# { "Depends": "py-genlayer:1jb45aa8ynh2a9c9xn3b7qqh8sm5q93hwfp7jqmwsfhh8jpz09h6" } # GenMint - AI-powered NFT platform with originality verification. # # Before an NFT is minted, validators independently fetch the content URL and # analyze it with an LLM to decide whether the work is original. Only work that # reaches consensus on originality gets minted. Listing prices are additionally # sanity-checked by an LLM fairness assessment. All comments are ASCII only. from genlayer import * import json from dataclasses import dataclass # --------------------------------------------------------------------------- # Error classification prefixes. # # [EXPECTED] - deterministic business logic error, validators must match exactly # [EXTERNAL] - deterministic external 4xx error, validators must match exactly # [TRANSIENT] - non-deterministic network / 5xx error, agree if both transient # [LLM_ERROR] - LLM misbehavior, always disagree to force validator rotation # --------------------------------------------------------------------------- ERROR_EXPECTED = "[EXPECTED]" ERROR_EXTERNAL = "[EXTERNAL]" ERROR_TRANSIENT = "[TRANSIENT]" ERROR_LLM = "[LLM_ERROR]" # Economic and scoring constants. BPS_DENOM = 10000 # basis-point denominator (100.00%) MIN_ORIGINALITY_BPS = 3000 # reject mints below 30.00% originality PLATFORM_FEE_BPS = 250 # 2.50% marketplace fee on sales MAX_CONTENT_CHARS = 4000 # cap content passed to the LLM VALID_RISK = ("LOW", "MEDIUM", "HIGH") # =========================================================================== # Module-level helpers (deterministic, no storage access). # =========================================================================== def _handle_leader_error(leaders_res, leader_fn) -> bool: """Compare a leader error against a fresh validator run. Deterministic errors must match exactly. Transient errors agree if both sides hit a transient failure. LLM or unknown errors disagree to force consensus rotation. """ leader_msg = getattr(leaders_res, "message", "") or "" try: leader_fn() # Leader errored but validator succeeded: disagree. return False except gl.vm.UserError as e: validator_msg = getattr(e, "message", "") or str(e) if validator_msg.startswith(ERROR_EXPECTED) or validator_msg.startswith(ERROR_EXTERNAL): return validator_msg == leader_msg if validator_msg.startswith(ERROR_TRANSIENT) and leader_msg.startswith(ERROR_TRANSIENT): return True return False except Exception: return False def _clean_json(text: str) -> dict: """Best-effort extraction of a JSON object from raw LLM text.""" import re first = text.find("{") last = text.rfind("}") if first == -1 or last == -1 or last < first: raise gl.vm.UserError(f"{ERROR_LLM} No JSON object in LLM output") body = text[first:last + 1] # Drop trailing commas that break strict json parsing. body = re.sub(r",(?!\s*?[\{\[\"\'\w])", "", body) try: parsed = json.loads(body) except Exception: raise gl.vm.UserError(f"{ERROR_LLM} Malformed JSON from LLM") if not isinstance(parsed, dict): raise gl.vm.UserError(f"{ERROR_LLM} LLM JSON is not an object") return parsed def _coerce_dict(resp) -> dict: """Normalize an exec_prompt result into a dict.""" if isinstance(resp, dict): return resp if isinstance(resp, str): return _clean_json(resp) raise gl.vm.UserError(f"{ERROR_LLM} Unexpected LLM response type: {type(resp)}") def _pick(data: dict, keys) -> object: """Return the first present key from a list of aliases, else None.""" for k in keys: if k in data and data[k] is not None: return data[k] return None def _parse_llm_int(data: dict, keys) -> int: raw = _pick(data, keys) if raw is None: raise gl.vm.UserError(f"{ERROR_LLM} Missing numeric field {keys}; keys={list(data.keys())}") try: return int(round(float(str(raw).strip()))) except (ValueError, TypeError): raise gl.vm.UserError(f"{ERROR_LLM} Non-numeric value: {raw}") def _parse_llm_bool(data: dict, keys) -> bool: raw = _pick(data, keys) if raw is None: raise gl.vm.UserError(f"{ERROR_LLM} Missing boolean field {keys}; keys={list(data.keys())}") if isinstance(raw, bool): return raw if isinstance(raw, (int, float)): return raw != 0 token = str(raw).strip().lower() if token in ("true", "yes", "y", "1", "original"): return True if token in ("false", "no", "n", "0", "plagiarized", "copied"): return False raise gl.vm.UserError(f"{ERROR_LLM} Non-boolean value: {raw}") def _normalize_risk(raw) -> str: """Map a free-form risk label onto LOW / MEDIUM / HIGH.""" if raw is None: raise gl.vm.UserError(f"{ERROR_LLM} Missing risk_level") token = str(raw).strip().upper() if token in VALID_RISK: return token if "HIGH" in token or "SEVERE" in token: return "HIGH" if "MED" in token or "MODERATE" in token: return "MEDIUM" if "LOW" in token or "MINIMAL" in token or "NONE" in token: return "LOW" raise gl.vm.UserError(f"{ERROR_LLM} Unrecognized risk_level: {raw}") def _clamp_bps(value: int) -> int: if value < 0: return 0 if value > BPS_DENOM: return BPS_DENOM return value def _safe_text(body) -> str: """Decode a web body to a bounded, LLM-safe text snippet.""" if body is None: return "" if isinstance(body, str): text = body else: try: text = bytes(body).decode("utf-8", errors="replace") except Exception: text = "" if len(text) > MAX_CONTENT_CHARS: text = text[:MAX_CONTENT_CHARS] return text def _now_iso() -> str: """Transaction datetime as an ISO string, best-effort.""" try: return str(gl.message_raw["datetime"]) except Exception: return "" # =========================================================================== # Storage dataclasses (flat - no nested collections). # =========================================================================== @allow_storage @dataclass class NFT: token_id: u256 creator: Address owner: Address title: str description: str content_url: str originality_score: u256 # basis points, 0..10000 is_verified: bool is_listed: bool price_atto: u256 # atto-scale (value * 10^18) created_at: str # ISO datetime string sold_count: u256 metadata_url: str # Advisory price-fairness verdict (set on list_nft, never blocks a listing). # A frontend can show a "price may be high" warning when is_price_fair is False. is_price_fair: bool price_fairness_reason: str @allow_storage @dataclass class CreatorProfile: address: Address total_minted: u256 total_sold: u256 reputation_score: u256 # basis points, weighted average originality is_banned: bool # =========================================================================== # Contract. # =========================================================================== class GenMint(gl.Contract): # --- storage fields (class-level annotations declare storage slots) --- owner: Address nfts: TreeMap[u256, NFT] creators: TreeMap[str, CreatorProfile] token_by_creator: TreeMap[str, u256] # key: "address_hex:index" token_count_by_creator: TreeMap[str, u256] # key: address_hex claimable: TreeMap[str, u256] # pull-payment ledger, key: address_hex next_token_id: u256 mint_fee_atto: u256 platform_earnings: u256 total_volume: u256 paused: bool def __init__(self, mint_fee_atto: u256): # NOTE: GenVM does not support default values on constructor args, so # mint_fee_atto is required. Pass 0 for a free mint. self.owner = gl.message.sender_address self.next_token_id = u256(1) self.mint_fee_atto = u256(mint_fee_atto) self.platform_earnings = u256(0) self.total_volume = u256(0) self.paused = False # ----------------------------------------------------------------- # Internal guards / storage helpers (deterministic). # ----------------------------------------------------------------- def _only_owner(self) -> None: if gl.message.sender_address != self.owner: raise gl.vm.UserError(f"{ERROR_EXPECTED} Owner only") def _require_not_paused(self) -> None: if self.paused: raise gl.vm.UserError(f"{ERROR_EXPECTED} Contract is paused") def _require_nft(self, token_id: int) -> NFT: if token_id not in self.nfts: raise gl.vm.UserError(f"{ERROR_EXPECTED} NFT does not exist") return self.nfts[token_id] def _touch_creator(self, hexkey: str, addr: Address) -> CreatorProfile: """Return the creator profile, creating an empty one if needed.""" if hexkey not in self.creators: self.creators[hexkey] = CreatorProfile( address=addr, total_minted=u256(0), total_sold=u256(0), reputation_score=u256(0), is_banned=False, ) return self.creators[hexkey] # ----------------------------------------------------------------- # Non-deterministic block 1: originality verification. # ----------------------------------------------------------------- def _check_originality(self, title: str, description: str, content_url: str) -> dict: def leader_fn(): resp = gl.nondet.web.request(content_url, method="GET") status = int(resp.status) if status >= 500: raise gl.vm.UserError(f"{ERROR_TRANSIENT} Content host returned {status}") if status >= 400: raise gl.vm.UserError(f"{ERROR_EXTERNAL} Content URL returned {status}") snippet = _safe_text(resp.body) prompt = ( "You are an originality auditor for an NFT marketplace. " "Assess whether the submitted artwork/content appears original " "or plagiarized based on the fetched content and its metadata.\n\n" "TITLE: " + str(title) + "\n" "DESCRIPTION: " + str(description) + "\n" "CONTENT_URL: " + str(content_url) + "\n" "FETCHED_CONTENT (truncated):\n" + snippet + "\n\n" "Return a JSON object with exactly these fields:\n" '{"originality_score": , ' '"is_original": , ' '"risk_level": <"LOW"|"MEDIUM"|"HIGH">}' ) analysis = gl.nondet.exec_prompt(prompt, response_format="json") data = _coerce_dict(analysis) score = _clamp_bps(_parse_llm_int( data, ("originality_score", "score", "originality", "rating"))) is_original = _parse_llm_bool(data, ("is_original", "original", "authentic")) risk = _normalize_risk(_pick(data, ("risk_level", "risk", "risk_rating"))) return { "originality_score": score, "is_original": is_original, "risk_level": risk, } def validator_fn(leaders_res: gl.vm.Result) -> bool: if not isinstance(leaders_res, gl.vm.Return): return _handle_leader_error(leaders_res, leader_fn) leader = leaders_res.calldata mine = leader_fn() # Agree on the categorical verdict; exact score match not required. same_risk = str(leader.get("risk_level")) == mine["risk_level"] same_verdict = bool(leader.get("is_original")) == bool(mine["is_original"]) return same_risk and same_verdict return gl.vm.run_nondet_unsafe(leader_fn, validator_fn) # ----------------------------------------------------------------- # Non-deterministic block 2: price fairness (no web fetch). # ----------------------------------------------------------------- def _assess_price_fairness( self, title: str, description: str, originality_score: int, price_atto: int ) -> dict: def leader_fn(): prompt = ( "You assess whether an NFT listing price is fair given its " "quality signals. Price is expressed in atto units " "(1 token = 10^18 atto).\n\n" "TITLE: " + str(title) + "\n" "DESCRIPTION: " + str(description) + "\n" "ORIGINALITY_SCORE_BPS: " + str(int(originality_score)) + "\n" "PRICE_ATTO: " + str(int(price_atto)) + "\n\n" "Return a JSON object with exactly these fields:\n" '{"is_fair": , "reason": }' ) analysis = gl.nondet.exec_prompt(prompt, response_format="json") data = _coerce_dict(analysis) is_fair = _parse_llm_bool(data, ("is_fair", "fair", "acceptable")) reason = str(_pick(data, ("reason", "explanation", "rationale")) or "") return {"is_fair": is_fair, "reason": reason} def validator_fn(leaders_res: gl.vm.Result) -> bool: if not isinstance(leaders_res, gl.vm.Return): return _handle_leader_error(leaders_res, leader_fn) leader = leaders_res.calldata mine = leader_fn() return bool(leader.get("is_fair")) == bool(mine["is_fair"]) return gl.vm.run_nondet_unsafe(leader_fn, validator_fn) # ================================================================= # Write methods. # ================================================================= @gl.public.write.payable def mint_nft(self, title: str, description: str, content_url: str, metadata_url: str) -> int: self._require_not_paused() sender = gl.message.sender_address hexkey = sender.as_hex profile = self._touch_creator(hexkey, sender) if profile.is_banned: raise gl.vm.UserError(f"{ERROR_EXPECTED} Creator is banned") if not title or not title.strip(): raise gl.vm.UserError(f"{ERROR_EXPECTED} Title is required") if not content_url or not content_url.lower().startswith(("http://", "https://")): raise gl.vm.UserError(f"{ERROR_EXPECTED} Valid content_url is required") if int(gl.message.value) < int(self.mint_fee_atto): raise gl.vm.UserError(f"{ERROR_EXPECTED} Insufficient mint fee") # Non-deterministic originality check runs before any storage writes. result = self._check_originality(title, description, content_url) score = _clamp_bps(int(result["originality_score"])) is_original = bool(result["is_original"]) if not is_original or score < MIN_ORIGINALITY_BPS: raise gl.vm.UserError( f"{ERROR_EXPECTED} Originality check failed (score={score}, original={is_original})" ) # --- storage writes (deterministic) --- token_id = int(self.next_token_id) self.nfts[u256(token_id)] = NFT( token_id=u256(token_id), creator=sender, owner=sender, title=title, description=description, content_url=content_url, originality_score=u256(score), is_verified=True, is_listed=False, price_atto=u256(0), created_at=_now_iso(), sold_count=u256(0), metadata_url=metadata_url, is_price_fair=True, # advisory default until listed price_fairness_reason="", ) # Weighted-average reputation using the count prior to this mint. prior = int(profile.total_minted) old_rep = int(profile.reputation_score) new_rep = (old_rep * prior + score) // (prior + 1) profile.reputation_score = u256(new_rep) profile.total_minted = u256(prior + 1) # Per-creator token index. idx = int(self.token_count_by_creator.get(hexkey, u256(0))) self.token_by_creator[hexkey + ":" + str(idx)] = u256(token_id) self.token_count_by_creator[hexkey] = u256(idx + 1) self.platform_earnings = u256(int(self.platform_earnings) + int(self.mint_fee_atto)) self.next_token_id = u256(token_id + 1) return token_id @gl.public.write def list_nft(self, token_id: int, price_atto: int) -> None: self._require_not_paused() nft = self._require_nft(token_id) if gl.message.sender_address != nft.owner: raise gl.vm.UserError(f"{ERROR_EXPECTED} Only owner can list") if not nft.is_verified: raise gl.vm.UserError(f"{ERROR_EXPECTED} NFT is not verified") if int(price_atto) <= 0: raise gl.vm.UserError(f"{ERROR_EXPECTED} Price must be positive") # Advisory only: record the fairness verdict but always allow the listing. assessment = self._assess_price_fairness( nft.title, nft.description, int(nft.originality_score), int(price_atto) ) nft.price_atto = u256(int(price_atto)) nft.is_listed = True nft.is_price_fair = bool(assessment["is_fair"]) nft.price_fairness_reason = str(assessment.get("reason", "")) @gl.public.write.payable def buy_nft(self, token_id: int) -> None: self._require_not_paused() nft = self._require_nft(token_id) if not nft.is_listed: raise gl.vm.UserError(f"{ERROR_EXPECTED} NFT is not listed") buyer = gl.message.sender_address seller = nft.owner if buyer == seller: raise gl.vm.UserError(f"{ERROR_EXPECTED} Cannot buy your own NFT") price = int(nft.price_atto) if int(gl.message.value) < price: raise gl.vm.UserError(f"{ERROR_EXPECTED} Insufficient payment") fee = (price * PLATFORM_FEE_BPS) // BPS_DENOM seller_amount = price - fee seller_key = seller.as_hex self.claimable[seller_key] = u256(int(self.claimable.get(seller_key, u256(0))) + seller_amount) self.platform_earnings = u256(int(self.platform_earnings) + fee) self.total_volume = u256(int(self.total_volume) + price) # Transfer ownership and clear the listing. nft.owner = buyer nft.is_listed = False nft.sold_count = u256(int(nft.sold_count) + 1) seller_profile = self._touch_creator(seller_key, seller) seller_profile.total_sold = u256(int(seller_profile.total_sold) + 1) @gl.public.write def delist_nft(self, token_id: int) -> None: nft = self._require_nft(token_id) if gl.message.sender_address != nft.owner: raise gl.vm.UserError(f"{ERROR_EXPECTED} Only owner can delist") nft.is_listed = False @gl.public.write def claim_earnings(self) -> int: sender = gl.message.sender_address hexkey = sender.as_hex amount = int(self.claimable.get(hexkey, u256(0))) if amount <= 0: raise gl.vm.UserError(f"{ERROR_EXPECTED} Nothing to claim") # Effects before interaction (pull-payment): zero the ledger first. self.claimable[hexkey] = u256(0) gl.get_contract_at(sender).emit_transfer(value=u256(amount), on="finalized") return amount @gl.public.write def set_mint_fee(self, fee: int) -> None: self._only_owner() if int(fee) < 0: raise gl.vm.UserError(f"{ERROR_EXPECTED} Fee must be non-negative") self.mint_fee_atto = u256(int(fee)) @gl.public.write def set_paused(self, value: bool) -> None: self._only_owner() self.paused = bool(value) @gl.public.write def ban_creator(self, address: str) -> None: self._only_owner() addr = Address(address) profile = self._touch_creator(addr.as_hex, addr) profile.is_banned = True # ================================================================= # View methods. # ================================================================= def _nft_to_dict(self, nft: NFT) -> dict: return { "token_id": int(nft.token_id), "creator": nft.creator.as_hex, "owner": nft.owner.as_hex, "title": nft.title, "description": nft.description, "content_url": nft.content_url, "originality_score": int(nft.originality_score), "is_verified": bool(nft.is_verified), "is_listed": bool(nft.is_listed), "price_atto": int(nft.price_atto), "created_at": nft.created_at, "sold_count": int(nft.sold_count), "metadata_url": nft.metadata_url, "is_price_fair": bool(nft.is_price_fair), "price_fairness_reason": nft.price_fairness_reason, } @gl.public.view def get_nft(self, token_id: int) -> dict: nft = self._require_nft(token_id) return self._nft_to_dict(nft) @gl.public.view def get_creator_profile(self, address: str) -> dict: hexkey = Address(address).as_hex if hexkey not in self.creators: return { "address": hexkey, "total_minted": 0, "total_sold": 0, "reputation_score": 0, "is_banned": False, } profile = self.creators[hexkey] return { "address": profile.address.as_hex, "total_minted": int(profile.total_minted), "total_sold": int(profile.total_sold), "reputation_score": int(profile.reputation_score), "is_banned": bool(profile.is_banned), } @gl.public.view def get_creator_tokens(self, address: str, offset: int, limit: int) -> list: hexkey = Address(address).as_hex count = int(self.token_count_by_creator.get(hexkey, u256(0))) out = [] start = max(0, int(offset)) end = min(count, start + max(0, int(limit))) i = start while i < end: token_id = int(self.token_by_creator[hexkey + ":" + str(i)]) if token_id in self.nfts: out.append(self._nft_to_dict(self.nfts[token_id])) i += 1 return out @gl.public.view def get_listed_nfts(self, offset: int, limit: int) -> list: total = int(self.next_token_id) out = [] skipped = 0 want = max(0, int(limit)) start = max(0, int(offset)) tid = 1 while tid < total and len(out) < want: if tid in self.nfts: nft = self.nfts[tid] if bool(nft.is_listed): if skipped < start: skipped += 1 else: out.append(self._nft_to_dict(nft)) tid += 1 return out @gl.public.view def get_platform_stats(self) -> dict: return { "total_nfts": int(self.next_token_id) - 1, "mint_fee_atto": int(self.mint_fee_atto), "platform_earnings": int(self.platform_earnings), "total_volume": int(self.total_volume), "paused": bool(self.paused), } @gl.public.view def get_claimable(self, address: str) -> int: hexkey = Address(address).as_hex return int(self.claimable.get(hexkey, u256(0)))