# v0.2.16 # { "Depends": "py-genlayer:1jb45aa8ynh2a9c9xn3b7qqh8sm5q93hwfp7jqmwsfhh8jpz09h6" } from genlayer import * import json import re from datetime import datetime # ─── Constants ──────────────────────────────────────────────────────────────── VERDICT_HEALTHY = "healthy" VERDICT_DEGRADED = "degraded" VERDICT_STALE = "stale" VERDICT_MISLEADING = "misleading" VERDICT_COMPROMISED = "compromised" VERDICT_FAILED = "failed" ALL_VERDICTS = [ VERDICT_HEALTHY, VERDICT_DEGRADED, VERDICT_STALE, VERDICT_MISLEADING, VERDICT_COMPROMISED, VERDICT_FAILED, ] STATUS_ACTIVE = "active" STATUS_CHALLENGED = "challenged" STATUS_RESOLVED = "resolved" STATUS_APPEALED = "appealed" STATUS_CLOSED = "closed" SIDE_SUPPORTER = "supporter" SIDE_CHALLENGER = "challenger" CATEGORY_WEATHER = "weather" CATEGORY_TVL = "tvl" CATEGORY_DOCS = "docs" CATEGORY_GOV = "gov" CATEGORY_AI = "ai" CATEGORY_PRICE = "price" CATEGORY_OTHER = "other" ALL_CATEGORIES = [ CATEGORY_WEATHER, CATEGORY_TVL, CATEGORY_DOCS, CATEGORY_GOV, CATEGORY_AI, CATEGORY_PRICE, CATEGORY_OTHER, ] # Minimum declared stake in units (1 unit = 1 GEN represented as integer) MIN_STAKE_UNITS: int = 10 # GEN's native denomination is wei; 1 GEN = 10**18 wei. All stake_units in this # contract are whole-GEN counts — actual escrowed value is stake_units * WEI_PER_GEN. WEI_PER_GEN: int = 10 ** 18 # Protocol fee: 200 basis points = 2% PROTOCOL_FEE_BPS: int = 200 # Claim lifetime before expiry (seconds) CLAIM_TTL_SECONDS: int = 30 * 24 * 3600 # Appeal window after first resolution (seconds) APPEAL_WINDOW_SECONDS: int = 3 * 24 * 3600 # Maximum secondary sources fetched per validation MAX_SECONDARY_SOURCES: int = 3 # Maximum stake-holders tracked per claim MAX_SUPPORTERS_PER_CLAIM: int = 50 # ─── Ghost interface for sending native GEN to EOAs ────────────────────────── # GenLayer requires EVM contract-interface syntax to send value even to plain # wallet addresses; this class has no real ABI, it only exists to make # emit_transfer() callable against an arbitrary Address. @gl.evm.contract_interface class _Payee: class View: pass class Write: pass # ─── Helper: safe TreeMap reads ────────────────────────────────────────────── def _rint(tmap, key, default: int = 0) -> int: """Read an int from a GenLayer TreeMap; return default if key is absent or None.""" try: val = tmap[key] if val is None: return default return int(val) except Exception: return default def _rstr(tmap, key, default: str = "") -> str: """Read a str from a GenLayer TreeMap; return default if key is absent or None.""" try: val = tmap[key] if val is None: return default return str(val) except Exception: return default # ─── Helper: JSON parsing ───────────────────────────────────────────────────── def _parse_json(text: str) -> dict: """ Robustly extract a JSON object from LLM output. Strips markdown fences, trailing commas, and surrounding noise. Returns {} on any parse failure. """ if isinstance(text, dict): return text raw = text.strip() # strip markdown code fences if raw.startswith("```"): lines = raw.split("\n") raw = "\n".join(lines[1:]) if raw.endswith("```"): raw = "\n".join(raw.split("\n")[:-1]) raw = raw.strip() # find outermost braces first = raw.find("{") last = raw.rfind("}") if first == -1 or last == -1 or last < first: return {} raw = raw[first : last + 1] # remove trailing commas before } or ] raw = re.sub(r",\s*([\}\]])", r"\1", raw) try: return json.loads(raw) except Exception: return {} def _extract_verdict(data: dict) -> str: """ Pull a valid verdict string from a parsed LLM result dict. Defaults to VERDICT_FAILED if missing or unrecognised. """ raw = str(data.get("verdict", "")).strip().lower() if raw in ALL_VERDICTS: return raw return VERDICT_FAILED def _extract_confidence(data: dict) -> int: """ Pull confidence as an integer 0-100 from a parsed result dict. Clamps to [0, 100]. """ try: val = float(data.get("confidence", 0.5)) # accept both 0-1 and 0-100 scale if val <= 1.0: val = val * 100.0 return max(0, min(100, int(round(val)))) except Exception: return 50 def _extract_reasoning(data: dict) -> str: """Pull reasoning string, truncated to 2000 chars.""" raw = str(data.get("reasoning", "No reasoning provided.")).strip() return raw[:2000] if len(raw) > 2000 else raw # ─── Contract ───────────────────────────────────────────────────────────────── class OracleRot(gl.Contract): """ OracleRot — the single on-chain registry for trust-monitoring claims. Storage layout (flat TreeMaps avoid @allow_storage complexity): Claims: claim_claimant : TreeMap[u256, Address] claim_source_url : TreeMap[u256, str] claim_assertion : TreeMap[u256, str] claim_category : TreeMap[u256, str] claim_status : TreeMap[u256, str] claim_verdict : TreeMap[u256, str] claim_confidence : TreeMap[u256, u256] # 0-100 claim_reasoning : TreeMap[u256, str] claim_created_at : TreeMap[u256, u256] claim_expires_at : TreeMap[u256, u256] claim_resolved_at : TreeMap[u256, u256] claim_appeal_at : TreeMap[u256, u256] # timestamp appeal was filed claim_finalized : TreeMap[u256, u256] # 1 once escrowed GEN has been paid out Stakes (supporter side, flat per-claim totals): claim_total_support : TreeMap[u256, u256] # total declared GEN units claim_supporter_count: TreeMap[u256, u256] Challenger: claim_challenger : TreeMap[u256, Address] claim_challenge_stake: TreeMap[u256, u256] claim_evidence_url : TreeMap[u256, str] claim_challenge_notes: TreeMap[u256, str] Individual supporter records (flat map keyed by "{claim_id}:{index}"): supporter_entry : TreeMap[str, str] (count tracked via claim_supporter_count above) Rewards accounting (flat map keyed by "{claim_id}:{index}"): reward_entry : TreeMap[str, str] claim_reward_count : TreeMap[u256, u256] Reputation: user_claims_filed : TreeMap[Address, u256] user_claims_won : TreeMap[Address, u256] # correct-side resolutions user_claims_lost : TreeMap[Address, u256] user_total_staked : TreeMap[Address, u256] user_total_earned : TreeMap[Address, u256] Global: next_claim_id : u256 treasury : Address total_protocol_fees : u256 total_claims_resolved: u256 owner : Address """ # ── Claim fields ────────────────────────────────────────────────────────── claim_claimant: TreeMap[u256, Address] claim_source_url: TreeMap[u256, str] claim_assertion: TreeMap[u256, str] claim_category: TreeMap[u256, str] claim_status: TreeMap[u256, str] claim_verdict: TreeMap[u256, str] claim_confidence: TreeMap[u256, u256] claim_reasoning: TreeMap[u256, str] claim_created_at: TreeMap[u256, u256] claim_expires_at: TreeMap[u256, u256] claim_resolved_at: TreeMap[u256, u256] claim_appeal_at: TreeMap[u256, u256] claim_finalized: TreeMap[u256, u256] # ── Stake / challenge fields ─────────────────────────────────────────────── claim_total_support: TreeMap[u256, u256] claim_supporter_count: TreeMap[u256, u256] claim_challenger: TreeMap[u256, Address] claim_challenge_stake: TreeMap[u256, u256] claim_evidence_url: TreeMap[u256, str] claim_challenge_notes: TreeMap[u256, str] supporter_entry: TreeMap[str, str] reward_entry: TreeMap[str, str] claim_reward_count: TreeMap[u256, u256] # ── Reputation ──────────────────────────────────────────────────────────── user_claims_filed: TreeMap[Address, u256] user_claims_won: TreeMap[Address, u256] user_claims_lost: TreeMap[Address, u256] user_total_staked: TreeMap[Address, u256] user_total_earned: TreeMap[Address, u256] # ── Global ──────────────────────────────────────────────────────────────── next_claim_id: u256 treasury: Address total_protocol_fees: u256 total_claims_resolved: u256 owner: Address # ───────────────────────────────────────────────────────────────────────── # Initialiser # ───────────────────────────────────────────────────────────────────────── def __init__(self) -> None: self.next_claim_id = u256(1) self.total_protocol_fees = u256(0) self.total_claims_resolved = u256(0) deployer = gl.message.sender_address self.treasury = deployer self.owner = deployer # ───────────────────────────────────────────────────────────────────────── # Write — Claim lifecycle # ───────────────────────────────────────────────────────────────────────── @gl.public.write.payable def register_claim( self, source_url: str, assertion: str, category: str, ) -> u256: """ Register a new trust-monitoring claim, escrowing the attached GEN as stake. Parameters ---------- source_url : The URL of the data source being monitored. assertion : Plain-English trust statement about the source. category : One of weather / tvl / docs / gov / ai / price / other. Stake is the GEN value attached to this transaction (≥ MIN_STAKE_UNITS GEN); it is held in the contract's balance until the claim resolves. Returns the new claim_id (u256). """ if not source_url or len(source_url.strip()) == 0: raise Exception("source_url cannot be empty") if not assertion or len(assertion.strip()) == 0: raise Exception("assertion cannot be empty") if category not in ALL_CATEGORIES: raise Exception( f"category must be one of: {', '.join(ALL_CATEGORIES)}" ) value = int(gl.message.value) if value < MIN_STAKE_UNITS * WEI_PER_GEN: raise Exception( f"stake must be at least {MIN_STAKE_UNITS} GEN" ) stake_units = value // WEI_PER_GEN sender = gl.message.sender_address now = u256(int(datetime.now().timestamp())) expires = u256(int(now) + CLAIM_TTL_SECONDS) cid = self.next_claim_id self.next_claim_id = u256(int(cid) + 1) self.claim_claimant[cid] = sender self.claim_source_url[cid] = source_url.strip() self.claim_assertion[cid] = assertion.strip() self.claim_category[cid] = category self.claim_status[cid] = STATUS_ACTIVE self.claim_verdict[cid] = "" self.claim_confidence[cid] = u256(0) self.claim_reasoning[cid] = "" self.claim_created_at[cid] = now self.claim_expires_at[cid] = expires self.claim_resolved_at[cid] = u256(0) self.claim_appeal_at[cid] = u256(0) self.claim_total_support[cid] = u256(stake_units) self.claim_supporter_count[cid] = u256(0) self.claim_reward_count[cid] = u256(0) self.claim_challenger[cid] = Address("0x0000000000000000000000000000000000000000") self.claim_challenge_stake[cid] = u256(0) self.claim_evidence_url[cid] = "" self.claim_challenge_notes[cid] = "" # record supporter entry as "address:amount" string entry = f"{str(sender)}:{stake_units}" self._supporter_append(cid, entry) # reputation prev_filed = _rint(self.user_claims_filed, sender) self.user_claims_filed[sender] = u256(prev_filed + 1) prev_staked = _rint(self.user_total_staked, sender) self.user_total_staked[sender] = u256(prev_staked + stake_units) return cid @gl.public.write.payable def support_claim(self, claim_id: u256) -> None: """ Add supporting stake to an existing active claim, escrowing the attached GEN (≥ 1 GEN) alongside the claimant's original stake. Parameters ---------- claim_id : The claim to support. """ self._require_claim_exists(claim_id) status = self.claim_status[claim_id] if status not in (STATUS_ACTIVE, STATUS_CHALLENGED): raise Exception("Claim is not open for additional staking") value = int(gl.message.value) if value < WEI_PER_GEN: raise Exception("stake must be at least 1 GEN") stake_units = value // WEI_PER_GEN sender = gl.message.sender_address count = _rint(self.claim_supporter_count, claim_id) if count >= MAX_SUPPORTERS_PER_CLAIM: raise Exception("Maximum supporter count reached for this claim") total = _rint(self.claim_total_support, claim_id) self.claim_total_support[claim_id] = u256(total + stake_units) entry = f"{str(sender)}:{stake_units}" self._supporter_append(claim_id, entry) prev_staked = _rint(self.user_total_staked, sender) self.user_total_staked[sender] = u256(prev_staked + stake_units) @gl.public.write.payable def challenge_claim( self, claim_id: u256, evidence_url: str, challenge_notes: str, ) -> None: """ Challenge a trust claim with counter-evidence, escrowing the attached GEN (≥ MIN_STAKE_UNITS) as the challenger's counter-stake. Parameters ---------- claim_id : The claim being challenged. evidence_url : URL hosting counter-evidence (can be empty string). challenge_notes : Human-readable reasoning for the challenge. """ self._require_claim_exists(claim_id) if self.claim_status[claim_id] != STATUS_ACTIVE: raise Exception("Only active claims can be challenged") sender = gl.message.sender_address claimant = self.claim_claimant[claim_id] if str(sender) == str(claimant): raise Exception("Claimant cannot challenge their own claim") value = int(gl.message.value) if value < MIN_STAKE_UNITS * WEI_PER_GEN: raise Exception( f"Challenge stake must be at least {MIN_STAKE_UNITS} GEN" ) stake_units = value // WEI_PER_GEN if not challenge_notes or len(challenge_notes.strip()) == 0: raise Exception("challenge_notes cannot be empty") self.claim_status[claim_id] = STATUS_CHALLENGED self.claim_challenger[claim_id] = sender self.claim_challenge_stake[claim_id] = u256(stake_units) self.claim_evidence_url[claim_id] = evidence_url.strip() if evidence_url else "" self.claim_challenge_notes[claim_id] = challenge_notes.strip() prev_staked = _rint(self.user_total_staked, sender) self.user_total_staked[sender] = u256(prev_staked + stake_units) @gl.public.write def resolve_claim(self, claim_id: u256) -> None: """ Trigger GenLayer validator consensus on a challenged or expired claim. This is the non-deterministic entry point. Each validator independently: 1. Fetches the primary source URL. 2. Fetches up to 3 secondary/reference sources. 3. Reasons about freshness, accuracy, integrity, and methodology. 4. Produces a structured verdict. Validators converge using eq_principle.prompt_comparative, requiring only the verdict label to match — confidence and reasoning may differ. This intentionally lenient consensus prevents UNDETERMINED outcomes. """ self._require_claim_exists(claim_id) sender = gl.message.sender_address claimant = self.claim_claimant[claim_id] status = self.claim_status[claim_id] now_ts = int(datetime.now().timestamp()) expires = _rint(self.claim_expires_at, claim_id) expired = now_ts >= expires if status == STATUS_ACTIVE: if not expired: raise Exception("Active claim has not expired yet — wait for the 30-day window") # expired+active without challenge: use close_expired_claim instead raise Exception("Use close_expired_claim for unchallenged expired claims") elif status == STATUS_CHALLENGED: # claimant can resolve anytime; anyone can trigger after expiry if str(sender) != str(claimant) and not expired: raise Exception("Only the claimant can resolve before the claim expires") elif status == STATUS_APPEALED: if str(sender) != str(claimant) and not expired: raise Exception("Only the claimant can resolve an appealed claim before expiry") else: raise Exception(f"Claim in status '{status}' cannot be resolved") # capture state for the non-deterministic closure source_url = self.claim_source_url[claim_id] assertion = self.claim_assertion[claim_id] category = self.claim_category[claim_id] evidence_url = self.claim_evidence_url[claim_id] challenge_notes = self.claim_challenge_notes[claim_id] # ── Non-deterministic validation ───────────────────────────────────── def validate_trust() -> str: return _run_validation( source_url=source_url, assertion=assertion, category=category, evidence_url=evidence_url, challenge_notes=challenge_notes, ) raw = gl.eq_principle.prompt_comparative( validate_trust, "The verdict field must be exactly one of: healthy, degraded, stale, " "misleading, compromised, failed. The verdict value must match exactly " "across all validators. Confidence and reasoning may differ freely.", ) data = _parse_json(raw) verdict = _extract_verdict(data) confidence = _extract_confidence(data) reasoning = _extract_reasoning(data) # persist resolution now_u = u256(now_ts) self.claim_status[claim_id] = STATUS_RESOLVED self.claim_verdict[claim_id] = verdict self.claim_confidence[claim_id] = u256(confidence) self.claim_reasoning[claim_id] = reasoning self.claim_resolved_at[claim_id] = now_u self.total_claims_resolved = u256(int(self.total_claims_resolved) + 1) # ── Escrowed GEN payout ────────────────────────────────────────────── # A claim resolved directly from CHALLENGED still has an open 72h appeal # window — escrowed GEN stays in the contract until that window elapses # (via finalize_claim) or the claimant appeals. A claim resolved from # APPEALED has already used its one allowed appeal, so this verdict is # final and the payout happens immediately. if status == STATUS_APPEALED: self._settle_rewards(claim_id, verdict) self.claim_finalized[claim_id] = u256(1) @gl.public.write def appeal_claim(self, claim_id: u256) -> None: """ File an appeal after a resolution. Only the claimant may appeal. One appeal per claim. Must be filed within APPEAL_WINDOW_SECONDS of resolution. Reopens the claim for a second non-deterministic resolution pass. """ self._require_claim_exists(claim_id) if self.claim_status[claim_id] != STATUS_RESOLVED: raise Exception("Only resolved claims can be appealed") sender = gl.message.sender_address claimant = self.claim_claimant[claim_id] if str(sender) != str(claimant): raise Exception("Only the original claimant can file an appeal") # check one-appeal-only: appeal_at > 0 means already appealed if _rint(self.claim_appeal_at, claim_id) > 0: raise Exception("This claim has already been appealed") if _rint(self.claim_finalized, claim_id) > 0: raise Exception("This claim has already been finalized and paid out") resolved_at = _rint(self.claim_resolved_at, claim_id) now_ts = int(datetime.now().timestamp()) if now_ts > resolved_at + APPEAL_WINDOW_SECONDS: raise Exception( f"Appeal window of {APPEAL_WINDOW_SECONDS // 3600}h has passed" ) # Escrowed GEN for a challenged/appealed claim is never paid out at the # first verdict — see resolve_claim/finalize_claim — precisely so an # appeal here has real funds left to redirect. Nothing to reverse. self.claim_status[claim_id] = STATUS_APPEALED self.claim_verdict[claim_id] = "" self.claim_confidence[claim_id]= u256(0) self.claim_reasoning[claim_id] = "" self.claim_appeal_at[claim_id] = u256(now_ts) @gl.public.write def finalize_claim(self, claim_id: u256) -> None: """ Release escrowed GEN for a resolved (challenged, never-appealed) claim once its 72-hour appeal window has elapsed. Anyone may call this — it only pays out a verdict that already stands and can no longer be appealed, so no special privilege is required. The backend's auto-expire scheduler calls this automatically so no manual action is needed for the payout to happen. """ self._require_claim_exists(claim_id) if self.claim_status[claim_id] != STATUS_RESOLVED: raise Exception("Claim is not in a resolved state") if _rint(self.claim_appeal_at, claim_id) > 0: raise Exception("This claim was appealed — funds settle via the appeal's final resolution") if _rint(self.claim_finalized, claim_id) > 0: raise Exception("Claim has already been finalized") resolved_at = _rint(self.claim_resolved_at, claim_id) now_ts = int(datetime.now().timestamp()) if now_ts <= resolved_at + APPEAL_WINDOW_SECONDS: raise Exception( f"Appeal window of {APPEAL_WINDOW_SECONDS // 3600}h has not elapsed yet" ) self._settle_rewards(claim_id, self.claim_verdict[claim_id]) self.claim_finalized[claim_id] = u256(1) @gl.public.write def close_expired_claim(self, claim_id: u256) -> None: """ Close an active claim that has passed its expiry without a challenge. Verdict defaults to HEALTHY (unchallenged = trusted). """ self._require_claim_exists(claim_id) if self.claim_status[claim_id] != STATUS_ACTIVE: raise Exception("Claim is not active") now_ts = int(datetime.now().timestamp()) if now_ts < _rint(self.claim_expires_at, claim_id): raise Exception("Claim has not expired yet") self.claim_status[claim_id] = STATUS_CLOSED self.claim_verdict[claim_id] = VERDICT_HEALTHY self.claim_confidence[claim_id] = u256(100) self.claim_reasoning[claim_id] = ( "Claim expired without a challenge. An unchallenged data source is " "presumed trustworthy by the protocol." ) self.claim_resolved_at[claim_id] = u256(now_ts) self.total_claims_resolved = u256(int(self.total_claims_resolved) + 1) self.claim_finalized[claim_id] = u256(1) # all escrowed stake returned to supporters self._return_stakes_to_supporters(claim_id) # ───────────────────────────────────────────────────────────────────────── # Write — Admin # ───────────────────────────────────────────────────────────────────────── @gl.public.write def set_treasury(self, new_treasury: Address) -> None: """Change the protocol treasury address. Owner only.""" self._require_owner() self.treasury = new_treasury @gl.public.write def transfer_ownership(self, new_owner: Address) -> None: """Transfer contract ownership. Owner only.""" self._require_owner() self.owner = new_owner # ───────────────────────────────────────────────────────────────────────── # View — Claims # ───────────────────────────────────────────────────────────────────────── @gl.public.view def get_claim(self, claim_id: u256) -> dict: """Return full claim details.""" self._require_claim_exists(claim_id) return { "claim_id": int(claim_id), "claimant": _rstr(self.claim_claimant, claim_id), "source_url": self.claim_source_url[claim_id], "assertion": self.claim_assertion[claim_id], "category": self.claim_category[claim_id], "status": self.claim_status[claim_id], "verdict": self.claim_verdict[claim_id], "confidence": _rint(self.claim_confidence, claim_id), "reasoning": self.claim_reasoning[claim_id], "total_support": _rint(self.claim_total_support, claim_id), "supporter_count": _rint(self.claim_supporter_count, claim_id), "challenger": _rstr(self.claim_challenger, claim_id), "challenge_stake": _rint(self.claim_challenge_stake, claim_id), "evidence_url": self.claim_evidence_url[claim_id], "challenge_notes": self.claim_challenge_notes[claim_id], "created_at": _rint(self.claim_created_at, claim_id), "expires_at": _rint(self.claim_expires_at, claim_id), "resolved_at": _rint(self.claim_resolved_at, claim_id), "appeal_at": _rint(self.claim_appeal_at, claim_id), "finalized": _rint(self.claim_finalized, claim_id) > 0, } @gl.public.view def get_claim_status(self, claim_id: u256) -> str: """Return the current status string for a claim.""" self._require_claim_exists(claim_id) return self.claim_status[claim_id] @gl.public.view def get_claim_verdict(self, claim_id: u256) -> str: """Return the verdict for a resolved claim (empty string if not yet resolved).""" self._require_claim_exists(claim_id) return self.claim_verdict[claim_id] @gl.public.view def get_claim_supporters(self, claim_id: u256) -> list: """Return raw supporter entries as ['address:amount', ...] strings.""" self._require_claim_exists(claim_id) return self._supporter_list(claim_id) @gl.public.view def get_claim_rewards(self, claim_id: u256) -> list: """Return reward distribution list as ['address:amount', ...] strings.""" self._require_claim_exists(claim_id) return self._reward_list(claim_id) @gl.public.view def get_total_claims(self) -> int: """Return the total number of claims registered.""" return int(self.next_claim_id) - 1 @gl.public.view def get_next_claim_id(self) -> int: """Return the ID that will be assigned to the next claim.""" return int(self.next_claim_id) # ───────────────────────────────────────────────────────────────────────── # View — User reputation # ───────────────────────────────────────────────────────────────────────── @gl.public.view def get_user_reputation(self, addr: Address) -> dict: """Return reputation stats for a wallet address.""" addr = Address(str(addr)) filed = _rint(self.user_claims_filed, addr) won = _rint(self.user_claims_won, addr) lost = _rint(self.user_claims_lost, addr) total_staked = _rint(self.user_total_staked, addr) total_earned = _rint(self.user_total_earned, addr) win_rate = 0 if filed > 0: win_rate = int((won * 100) // filed) return { "address": str(addr), "claims_filed":filed, "claims_won": won, "claims_lost": lost, "win_rate_pct":win_rate, "total_staked":total_staked, "total_earned":total_earned, "net_return": total_earned - total_staked, } # ───────────────────────────────────────────────────────────────────────── # View — Protocol stats # ───────────────────────────────────────────────────────────────────────── @gl.public.view def get_protocol_stats(self) -> dict: """Return global protocol statistics.""" return { "total_claims": int(self.next_claim_id) - 1, "total_resolved": int(self.total_claims_resolved), "total_fees": int(self.total_protocol_fees), "treasury": str(self.treasury), "owner": str(self.owner), "min_stake": MIN_STAKE_UNITS, "fee_bps": PROTOCOL_FEE_BPS, "claim_ttl_days": CLAIM_TTL_SECONDS // 86400, "appeal_window_h": APPEAL_WINDOW_SECONDS // 3600, } @gl.public.view def get_min_stake(self) -> int: """Return the minimum stake required to open a claim.""" return MIN_STAKE_UNITS @gl.public.view def can_appeal(self, claim_id: u256) -> bool: """Return True if the claim is currently within its appeal window.""" self._require_claim_exists(claim_id) if self.claim_status[claim_id] != STATUS_RESOLVED: return False if _rint(self.claim_appeal_at, claim_id) > 0: return False now_ts = int(datetime.now().timestamp()) resolved_at = _rint(self.claim_resolved_at, claim_id) return now_ts <= resolved_at + APPEAL_WINDOW_SECONDS @gl.public.view def can_finalize(self, claim_id: u256) -> bool: """Return True if a resolved claim's escrowed GEN is ready to be paid out.""" self._require_claim_exists(claim_id) if self.claim_status[claim_id] != STATUS_RESOLVED: return False if _rint(self.claim_appeal_at, claim_id) > 0: return False if _rint(self.claim_finalized, claim_id) > 0: return False now_ts = int(datetime.now().timestamp()) resolved_at = _rint(self.claim_resolved_at, claim_id) return now_ts > resolved_at + APPEAL_WINDOW_SECONDS @gl.public.view def is_expired(self, claim_id: u256) -> bool: """Return True if the claim has passed its expiry timestamp.""" self._require_claim_exists(claim_id) now_ts = int(datetime.now().timestamp()) return now_ts >= _rint(self.claim_expires_at, claim_id) # ───────────────────────────────────────────────────────────────────────── # Internal helpers # ───────────────────────────────────────────────────────────────────────── def _supporter_append(self, claim_id: u256, entry: str) -> None: """Append an entry to the supporter list and increment the count.""" count = _rint(self.claim_supporter_count, claim_id) self.supporter_entry[f"{int(claim_id)}:{count}"] = entry self.claim_supporter_count[claim_id] = u256(count + 1) def _supporter_list(self, claim_id: u256) -> list: """Return all supporter entries for a claim.""" count = _rint(self.claim_supporter_count, claim_id) return [self.supporter_entry[f"{int(claim_id)}:{i}"] for i in range(count)] def _reward_append(self, claim_id: u256, entry: str) -> None: """Append an entry to the reward list and increment the count.""" count = _rint(self.claim_reward_count, claim_id) self.reward_entry[f"{int(claim_id)}:{count}"] = entry self.claim_reward_count[claim_id] = u256(count + 1) def _reward_list(self, claim_id: u256) -> list: """Return all reward entries for a claim.""" count = _rint(self.claim_reward_count, claim_id) return [self.reward_entry[f"{int(claim_id)}:{i}"] for i in range(count)] def _require_claim_exists(self, claim_id: u256) -> None: if int(claim_id) < 1 or int(claim_id) >= int(self.next_claim_id): raise Exception(f"Claim {int(claim_id)} does not exist") def _require_owner(self) -> None: if str(gl.message.sender_address) != str(self.owner): raise Exception("Caller is not the contract owner") def _pay(self, addr_str: str, amount_units: int) -> None: """ Transfer `amount_units` GEN (whole units, converted to wei here) out of this contract's escrowed balance to an EOA/address. No-op for zero amounts or the null address. """ if amount_units <= 0: return null_addr = "0x0000000000000000000000000000000000000000" if addr_str == null_addr: return _Payee(Address(addr_str)).emit_transfer(value=u256(amount_units * WEI_PER_GEN)) def _settle_rewards(self, claim_id: u256, verdict: str) -> None: """ Pay out escrowed GEN for a finally-decided claim. healthy verdict → supporters win the challenge pot; challenger loses. any other verdict→ challenger wins the support pot; supporters lose. Protocol takes PROTOCOL_FEE_BPS of the losing side's stake, sent to the treasury. All amounts are recorded in the reward ledger for auditability, and reputation counters are updated on-chain. """ total_support = _rint(self.claim_total_support, claim_id) challenge_stake = _rint(self.claim_challenge_stake, claim_id) if verdict == VERDICT_HEALTHY: # supporters win — challenger's stake flows to supporters proportionally losing_side_total = challenge_stake fee = (losing_side_total * PROTOCOL_FEE_BPS) // 10000 net_winnings = losing_side_total - fee self.total_protocol_fees = u256( int(self.total_protocol_fees) + fee ) self._pay(str(self.treasury), fee) # distribute winnings proportionally across supporters if total_support > 0 and net_winnings > 0: supporter_entries = self._supporter_list(claim_id) for entry in supporter_entries: parts = entry.split(":") if len(parts) != 2: continue addr_str, amt_str = parts try: amt = int(amt_str) share = (amt * net_winnings) // total_support payout = amt + share reward_entry = f"{addr_str}:{payout}" self._reward_append(claim_id, reward_entry) self._pay(addr_str, payout) # update earned reputation addr = Address(addr_str) prev_earned = _rint(self.user_total_earned, addr) self.user_total_earned[addr] = u256(prev_earned + payout) prev_won = _rint(self.user_claims_won, addr) self.user_claims_won[addr] = u256(prev_won + 1) except Exception: continue # challenger lost challenger = self.claim_challenger[claim_id] null_addr = "0x0000000000000000000000000000000000000000" if str(challenger) != null_addr: prev_lost = _rint(self.user_claims_lost, challenger) self.user_claims_lost[challenger] = u256(prev_lost + 1) else: # challenger wins — supporters' stake flows to challenger losing_side_total = total_support fee = (losing_side_total * PROTOCOL_FEE_BPS) // 10000 net_winnings = losing_side_total - fee self.total_protocol_fees = u256( int(self.total_protocol_fees) + fee ) self._pay(str(self.treasury), fee) challenger = self.claim_challenger[claim_id] null_addr = "0x0000000000000000000000000000000000000000" if str(challenger) != null_addr: prize = int(challenge_stake) + net_winnings reward_entry = f"{str(challenger)}:{prize}" self._reward_append(claim_id, reward_entry) self._pay(str(challenger), prize) prev_earned = _rint(self.user_total_earned, challenger) self.user_total_earned[challenger] = u256(prev_earned + prize) prev_won = _rint(self.user_claims_won, challenger) self.user_claims_won[challenger] = u256(prev_won + 1) # supporters lost supporter_entries = self._supporter_list(claim_id) for entry in supporter_entries: parts = entry.split(":") if len(parts) != 2: continue addr_str, _ = parts try: addr = Address(addr_str) prev_lost = _rint(self.user_claims_lost, addr) self.user_claims_lost[addr] = u256(prev_lost + 1) except Exception: continue def _return_stakes_to_supporters(self, claim_id: u256) -> None: """Return all escrowed stakes to supporters (used for expired+unchallenged).""" supporter_entries = self._supporter_list(claim_id) for entry in supporter_entries: parts = entry.split(":") if len(parts) != 2: continue addr_str, amt_str = parts try: amt = int(amt_str) self._reward_append(claim_id, f"{addr_str}:{amt}") self._pay(addr_str, amt) addr = Address(addr_str) prev_earned = _rint(self.user_total_earned, addr) self.user_total_earned[addr] = u256(prev_earned + amt) prev_won = _rint(self.user_claims_won, addr) self.user_claims_won[addr] = u256(prev_won + 1) except Exception: continue # ───────────────────────────────────────────────────────────────────────────── # Non-deterministic validation logic (module-level, called inside closure) # ───────────────────────────────────────────────────────────────────────────── def _run_validation( source_url: str, assertion: str, category: str, evidence_url: str, challenge_notes: str, ) -> str: """ Perform independent multi-source trust validation. Steps: 1. Fetch the primary source URL. 2. Fetch relevant secondary sources based on category. 3. If a challenger provided evidence, fetch it too. 4. Submit everything to the LLM for structured verdict. 5. Return the raw JSON string for eq_principle comparison. Designed to always return one of the six verdict labels — the prompt explicitly forbids UNDETERMINED responses, making consensus robust. """ # ── Step 1: Fetch primary source ───────────────────────────────────────── primary_content = _safe_fetch(source_url) # ── Step 2: Fetch secondary reference sources ───────────────────────────── secondary_sources = _get_secondary_sources(source_url, category) secondary_content_blocks = [] for name, url in secondary_sources: content = _safe_fetch(url) if content: secondary_content_blocks.append( f"[Reference: {name} — {url}]\n{content[:1500]}" ) secondary_section = ( "\n\n".join(secondary_content_blocks) if secondary_content_blocks else "No secondary sources could be fetched." ) # ── Step 3: Fetch challenger evidence if provided ───────────────────────── evidence_section = "" if evidence_url and evidence_url.strip(): evidence_content = _safe_fetch(evidence_url) evidence_section = ( f"\n## Challenger Evidence URL: {evidence_url}\n" f"{evidence_content[:2000] if evidence_content else '[Evidence URL unreachable]'}\n" f"\n## Challenger's Stated Reason\n{challenge_notes}\n" ) # ── Step 4: Build the validation prompt ─────────────────────────────────── category_guidance = _category_guidance(category) prompt = f"""You are an independent AI validator for the OracleRot trust-monitoring protocol. Your task is to determine whether a data source remains trustworthy based on current evidence. ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ CLAIM UNDER REVIEW ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ Source URL: {source_url} Category: {category} Trust Assertion: "{assertion}" ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ PRIMARY SOURCE CONTENT (fetched live) ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ {primary_content[:3000] if primary_content else "[PRIMARY SOURCE UNREACHABLE]"} ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ SECONDARY REFERENCE SOURCES ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ {secondary_section} {evidence_section} ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ CATEGORY-SPECIFIC VALIDATION GUIDANCE ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ {category_guidance} ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ VALIDATION DIMENSIONS — assess each: ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ 1. FRESHNESS — Is the data being updated at the expected frequency for this category? 2. ACCURACY — Do values appear internally consistent and plausible? 3. REACHABILITY — Is the source accessible and responding with valid content? 4. INTEGRITY — Any evidence of selective omission, manipulation, or misleading framing? 5. METHODOLOGY — Has the underlying methodology changed in ways that break the assertion? 6. CROSS-SOURCE AGREEMENT — Does primary content agree with independent references? ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ VERDICT OPTIONS — choose exactly ONE: ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ healthy → Source functions correctly; assertion remains fully valid. degraded → Source partially works but quality, freshness, or completeness has declined. stale → Data exists but is no longer updated at expected frequency. misleading → Data is present but selectively presented or framed to distort understanding. compromised → Evidence of external manipulation, injection, or corruption of the source. failed → Source has materially failed; the trust assertion is no longer true. IMPORTANT RULES: - You MUST choose exactly one of the six verdicts above. No other values are permitted. - Do NOT output "undetermined", "unknown", "N/A", or any other value. - If evidence is ambiguous or limited, make the most reasonable determination from available data. - If the source is unreachable, default to "stale" for expected-live sources or "failed" for critical feeds. - If a challenger provided evidence, weigh it seriously but do not automatically rule in their favour. - Lean toward "healthy" only if you have positive evidence the source is working correctly. - Lean toward "degraded" for marginal cases rather than extreme verdicts. ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ REQUIRED OUTPUT FORMAT ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ Respond with ONLY valid JSON — no markdown, no preamble, no trailing text: {{ "verdict": "", "confidence": , "freshness_score": , "accuracy_score": , "reachability_score": , "integrity_score": , "reasoning": "<2-5 sentence explanation referencing the evidence you examined>" }}""" result = gl.nondet.exec_prompt(prompt) return result def _safe_fetch(url: str) -> str: """ Fetch a URL using GenLayer's web access tools. Tries gl.nondet.web.get() first for raw HTTP (best for APIs/JSON), then falls back to gl.nondet.web.render() for JS-heavy pages. Returns an empty string on any failure. """ if not url or not url.strip(): return "" url = url.strip() # Try raw HTTP GET first — fast and reliable for APIs and static content try: resp = gl.nondet.web.get(url) if resp and len(resp) > 0: return resp except Exception: pass # Fall back to full page render for JS-heavy or dynamic pages try: content = gl.nondet.web.render( url, mode="text", wait_after_loaded="5s", ) return content or "" except Exception: return "" def _get_secondary_sources(primary_url: str, category: str) -> list: """ Return a list of (label, url) tuples for secondary reference sources based on the claim category. These are well-known, stable reference URLs. """ sources = [] if category == CATEGORY_WEATHER: sources = [ ("NOAA Climate", "https://www.ncei.noaa.gov/access/monitoring/monthly-report/"), ("Weather Underground History", "https://www.wunderground.com/history"), ("Open-Meteo Docs", "https://open-meteo.com/en/docs"), ] elif category == CATEGORY_TVL: sources = [ ("DeFiLlama TVL", "https://defillama.com/"), ("DeFiLlama API Docs", "https://defillama.com/docs/api"), ] elif category == CATEGORY_DOCS: sources = [ ("Wayback Machine CDX API", f"https://web.archive.org/cdx/search/cdx?url={primary_url}&output=json&limit=5"), ] elif category == CATEGORY_GOV: sources = [ ("Data.gov Catalog", "https://catalog.data.gov/dataset"), ("EU Open Data Portal", "https://data.europa.eu/en"), ] elif category == CATEGORY_AI: sources = [ ("Hugging Face Models", "https://huggingface.co/models"), ("Papers With Code", "https://paperswithcode.com/sota"), ] elif category == CATEGORY_PRICE: sources = [ ("CoinGecko API", "https://api.coingecko.com/api/v3/ping"), ("CoinMarketCap", "https://coinmarketcap.com/"), ] else: # generic: try archive and a search context sources = [ ("Wayback Machine", f"https://web.archive.org/web/*/{primary_url}"), ] return sources[:MAX_SECONDARY_SOURCES] def _category_guidance(category: str) -> str: """Return category-specific validation instructions for the LLM.""" guidance = { CATEGORY_WEATHER: ( "Weather APIs: check that the most recent observation timestamp is within " "the last 1-6 hours for real-time feeds, or within the expected update cycle. " "Verify values (temperature, humidity, wind) are within physically plausible ranges. " "Compare with nearby NOAA or WMO stations if available." ), CATEGORY_TVL: ( "DeFi TVL feeds: check that the reported TVL is being updated at least daily. " "Compare figures across DeFiLlama and any on-chain data. " "Flag if TVL has dropped >50% without a corresponding market event as potential manipulation. " "Check that token prices used in TVL calculation appear current." ), CATEGORY_DOCS: ( "Documentation sources: check that the described API/protocol behavior matches " "the current implementation. Look for 'last updated' timestamps. " "Flag if changelog entries contradict the documented behavior. " "Compare with any versioned changelog or GitHub commit history if accessible." ), CATEGORY_GOV: ( "Government datasets: check the 'last updated' or 'data through' field. " "For weekly datasets, flag as stale if >10 days since last update. " "For monthly datasets, flag as stale if >45 days since last update. " "Check that field definitions and methodology notes have not changed." ), CATEGORY_AI: ( "AI/ML endpoints: check that the API responds with expected output format. " "Look for deprecation notices, model version changes, or accuracy regressions. " "Check that benchmark scores or documentation have not been silently revised. " "Verify response latency is within expected parameters." ), CATEGORY_PRICE: ( "Price feeds: check that prices are updating at expected frequency (CEX: seconds, " "DEX: blocks). Compare spot prices across at least two independent sources. " "Flag spreads >2% between sources as potential manipulation. " "Check for circuit-breaker or halted state indicators." ), CATEGORY_OTHER: ( "General data source: apply standard trust-monitoring criteria. " "Check freshness relative to the source's stated update cadence. " "Compare values with any available independent cross-reference. " "Flag significant deviations from expected behaviour." ), } return guidance.get(category, guidance[CATEGORY_OTHER])