# { "Depends": "py-genlayer:1jb45aa8ynh2a9c9xn3b7qqh8sm5q93hwfp7jqmwsfhh8jpz09h6" } import json import re from dataclasses import dataclass from genlayer import * # ============================================================================= # ENTERPRISE PROTOCOL CONFIGURATION # ============================================================================= # These constants define the mathematical boundaries, context lengths, and # fault-tolerant limits for the underlying consensus protocol. BPS_DENOMINATOR = 10000 # 100.00% expressed in basis points MAX_SUMMARY_LEN = 2000 # Maximum allowed string length for user match summaries MAX_CONTEXT_LEN = 4000 # Maximum external context memory buffer PROTOCOL_VERSION = "v2.5.0-Enterprise" MAX_SYSTEM_METRICS = 50 # Limit for system metric iterations # ============================================================================= # ERROR CLASSIFICATION SCHEMA # ============================================================================= # GenLayer VM processes exceptions deterministically across validation nodes. # By strictly prefixing user errors, the consensus layer can gracefully handle # disagreements (e.g. differentiating an external 502 vs an LLM hallucination). ERROR_EXPECTED = "[EXPECTED]" # Deterministic business-logic violation ERROR_EXTERNAL = "[EXTERNAL]" # Deterministic external oracle failure ERROR_TRANSIENT = "[TRANSIENT]" # Non-deterministic network partition ERROR_LLM = "[LLM_ERROR]" # Consensus breakdown or prompt hallucination @allow_storage @dataclass class PlayerProfile: """ Core data structure representing a gamer profile. By utilizing strict dataclasses, we guarantee memory layout determinism across all nodes running GenVM, avoiding float/dict instantiation drifts. """ address: str handle: str total_kills: u256 total_wins: u256 matches_played: u256 reputation: u256 rank: str current_loadout: str faction: str lore: str achievements: str @allow_storage @dataclass class MatchRecord: """ Immutable ledger of a single match. """ match_id: u256 player: str difficulty: str kills: u256 won: bool weapon_used: str narrative: str achievement_title: str created_at: u256 @allow_storage @dataclass class BossEncounter: """ Data structure for global community events. """ boss_id: u256 boss_name: str hp: u256 current_hp: u256 lore: str status: str created_at: u256 @allow_storage @dataclass class ProtocolState: """ High-level encapsulation of the entire protocol's thermodynamic state. Used for mathematical divergence calculations and on-chain analytics. """ match_counter: u256 boss_counter: u256 total_players: u256 total_kills_global: u256 global_lore: str season: u256 global_risk_index_bps: u256 class GameLeaderboard(gl.Contract): """ GenLayer Royale Enterprise Architecture. This contract handles the ingestion, validation, and AI-driven curation of an entire cyberpunk game universe. It guarantees Byzantine fault tolerance by de-coupling deterministic state storage from non-deterministic LLM execution. """ profiles: TreeMap[str, PlayerProfile] matches: TreeMap[str, MatchRecord] bosses: TreeMap[str, BossEncounter] player_ids: DynArray[str] match_ids: DynArray[str] boss_ids: DynArray[str] state: ProtocolState def __init__(self): """ Initializes the state trees. TreeMaps provide O(log n) deterministic lookups, satisfying strict GenLayer performance constraints. """ self.state.match_counter = u256(0) self.state.boss_counter = u256(0) self.state.total_players = u256(0) self.state.total_kills_global = u256(0) self.state.global_lore = "The Neon City is under siege by the Megacorps. Only rogue hackers can breach the mainframe." self.state.season = u256(1) self.state.global_risk_index_bps = u256(0) # ------------------------------------------------------------------------- # CORE BUSINESS LOGIC # ------------------------------------------------------------------------- @gl.public.write def register_player(self, handle: str) -> bool: """ Registers a new player profile on the ledger. """ sender = str(gl.message.sender_address).lower() if sender in self.profiles: return False clean_handle = _deep_sanitize(handle)[:64] if not clean_handle: clean_handle = "Rookie_" + sender[:4] prof = PlayerProfile( address=sender, handle=clean_handle, total_kills=u256(0), total_wins=u256(0), matches_played=u256(0), reputation=u256(0), rank="Scrap Scavenger", current_loadout="Standard Pistol", faction="Unaligned", lore="A shadow on the network.", achievements="[]" ) self.profiles[sender] = prof self.player_ids.append(sender) self.state.total_players = u256(int(self.state.total_players) + 1) return True def _ensure_registered(self, sender: str): if sender not in self.profiles: prof = PlayerProfile( address=sender, handle="Rookie_" + sender[:4], total_kills=u256(0), total_wins=u256(0), matches_played=u256(0), reputation=u256(0), rank="Scrap Scavenger", current_loadout="Standard Pistol", faction="Unaligned", lore="A shadow on the network.", achievements="[]" ) self.profiles[sender] = prof self.player_ids.append(sender) self.state.total_players = u256(int(self.state.total_players) + 1) @gl.public.write def generate_player_lore(self) -> bool: """ Triggers the GenLayer Subjective Consensus engine to dynamically generate a unique player lore based on live real-world news feeds and their current loadout. """ sender = str(gl.message.sender_address).lower() self._ensure_registered(sender) prof = self.profiles[sender] def leader_fn() -> dict: tech_news = _fetch_url("https://news.ycombinator.com/") prompt = ( f"You are the AI Black Market Dealer.\\n" f"Tech news: {tech_news}\\n" f"Player loadout: {prof.current_loadout}\\n" f"Player rank: {prof.rank}\\n" "Generate a short, gritty Cyberpunk backstory/lore for this player.\\n" 'Return ONLY a JSON object: {"lore": ""}' ) analysis = gl.nondet.exec_prompt(prompt, response_format="json") return { "lore": _clean_string(analysis, "lore", 512, "A shadow on the network.") } def validator_fn(leader_res) -> bool: if not isinstance(leader_res, gl.vm.Return): return _handle_leader_error(leader_res, leader_fn) leader = leader_res.calldata if not isinstance(leader, dict): return False val_prompt = ( f"Review lore: '{leader.get('lore')}'.\\n" "Does it fit a gritty cyberpunk theme?\\n" "Output strictly YES or NO." ) val_output = gl.nondet.exec_prompt(val_prompt, response_format="text") if isinstance(val_output, str) and "YES" in val_output.upper(): return True return False decision = gl.vm.run_nondet_unsafe(leader_fn, validator_fn) prof.lore = decision["lore"] self.profiles[sender] = prof return True @gl.public.write def generate_cyberpunk_loadout(self) -> bool: """ Triggers the GenLayer Subjective Consensus engine to dynamically generate a unique loadout based on live real-world news feeds. """ sender = str(gl.message.sender_address).lower() self._ensure_registered(sender) prof = self.profiles[sender] def leader_fn() -> dict: tech_news = _fetch_url("https://news.ycombinator.com/") prompt = _interpret_loadout_prompt(tech_news) analysis = gl.nondet.exec_prompt(prompt, response_format="json") return { "weapon_name": _clean_string(analysis, "weapon_name", 64, "Rusty Pipe"), "special_effect": _clean_string(analysis, "special_effect", 256, "None") } def validator_fn(leader_res) -> bool: if not isinstance(leader_res, gl.vm.Return): return _handle_leader_error(leader_res, leader_fn) leader = leader_res.calldata if not isinstance(leader, dict): return False val_prompt = _interpret_loadout_val_prompt(leader) val_output = gl.nondet.exec_prompt(val_prompt, response_format="text") if isinstance(val_output, str) and "YES" in val_output.upper(): return True return False decision = gl.vm.run_nondet_unsafe(leader_fn, validator_fn) prof.current_loadout = decision["weapon_name"] self.profiles[sender] = prof return True @gl.public.write def record_match(self, difficulty: str, kills: int, summary: str) -> int: """ Logs a match on-chain and triggers an AI consensus engine to generate personalized lore and achievements based on performance. """ sender = str(gl.message.sender_address).lower() self._ensure_registered(sender) if kills < 0: raise gl.vm.UserError(f"{ERROR_EXPECTED} Invalid kills") clean_summary = _deep_sanitize(summary)[:MAX_SUMMARY_LEN] won = "won the match" in summary.lower() or "victory" in summary.lower() self.state.match_counter = u256(int(self.state.match_counter) + 1) m_id = int(self.state.match_counter) prof = self.profiles[sender] weapon = prof.current_loadout def leader_fn() -> dict: tech_news = _fetch_url("https://news.ycombinator.com/") prompt = _interpret_match_prompt(won, kills, difficulty, weapon, clean_summary, tech_news) analysis = gl.nondet.exec_prompt(prompt, response_format="json") return { "narrative": _clean_string(analysis, "narrative", 512, "A battle in the shadows."), "achievement_title": _clean_string(analysis, "achievement_title", 64, "Glitch"), "faction": _clean_string(analysis, "faction", 64, "Unaligned") } def validator_fn(leader_res) -> bool: if not isinstance(leader_res, gl.vm.Return): return _handle_leader_error(leader_res, leader_fn) leader = leader_res.calldata if not isinstance(leader, dict): return False val_prompt = _interpret_match_val_prompt(leader) val_output = gl.nondet.exec_prompt(val_prompt, response_format="text") if isinstance(val_output, str) and "YES" in val_output.upper(): return True return False decision = gl.vm.run_nondet_unsafe(leader_fn, validator_fn) # State transitions m_record = MatchRecord( match_id=u256(m_id), player=sender, difficulty=difficulty, kills=u256(kills), won=won, weapon_used=weapon, narrative=decision["narrative"], achievement_title=decision["achievement_title"], created_at=u256(0) ) self.matches[str(m_id)] = m_record self.match_ids.append(str(m_id)) prof.total_kills = u256(int(prof.total_kills) + kills) prof.total_wins = u256(int(prof.total_wins) + (1 if won else 0)) prof.matches_played = u256(int(prof.matches_played) + 1) prof.reputation = u256(int(prof.reputation) + (kills * (2 if won else 1))) achs = json.loads(prof.achievements) achs.append(decision["achievement_title"]) prof.achievements = json.dumps(achs) prof.faction = decision["faction"] rep = int(prof.reputation) if rep > 1000: prof.rank = "Night City Legend" elif rep > 500: prof.rank = "Elite Runner" elif rep > 100: prof.rank = "Street Samurai" self.profiles[sender] = prof self.state.total_kills_global = u256(int(self.state.total_kills_global) + kills) return m_id @gl.public.write def trigger_world_boss_event(self) -> bool: """ Triggers a global AI event that spawns a boss based on live macro news. """ self.state.boss_counter = u256(int(self.state.boss_counter) + 1) b_id = int(self.state.boss_counter) def leader_fn() -> dict: news = _fetch_url("https://www.coindesk.com/") prompt = _interpret_boss_prompt(news) analysis = gl.nondet.exec_prompt(prompt, response_format="json") return { "boss_name": _clean_string(analysis, "boss_name", 64, "The Glitch"), "hp": _parse_ratio_bps(analysis, "hp", 100000), "lore_description": _clean_string(analysis, "lore_description", 512, "A remnant of the old web.") } def validator_fn(leader_res) -> bool: if not isinstance(leader_res, gl.vm.Return): return _handle_leader_error(leader_res, leader_fn) leader = leader_res.calldata if not isinstance(leader, dict): return False val_prompt = _interpret_boss_val_prompt(leader) val_output = gl.nondet.exec_prompt(val_prompt, response_format="text") if isinstance(val_output, str) and "YES" in val_output.upper(): return True return False decision = gl.vm.run_nondet_unsafe(leader_fn, validator_fn) boss = BossEncounter( boss_id=u256(b_id), boss_name=decision["boss_name"], hp=u256(decision["hp"]), current_hp=u256(decision["hp"]), lore=decision["lore_description"], status="ALIVE", created_at=u256(0) ) self.bosses[str(b_id)] = boss self.boss_ids.append(str(b_id)) return True @gl.public.view def get_highest_score(self, difficulty: str) -> int: """Returns the highest kills achieved in a specific difficulty.""" highest = 0 total_matches = int(self.state.match_counter) for i in range(1, total_matches + 1): m_id = str(i) if m_id in self.matches: m = self.matches[m_id] if str(m.difficulty).upper() == difficulty.upper(): kills = int(m.kills) if kills > highest: highest = kills return highest @gl.public.view def get_match(self, match_id: int) -> str: if str(match_id) not in self.matches: return "{}" m = self.matches[str(match_id)] return json.dumps({ "match_id": int(m.match_id), "player": m.player, "difficulty": m.difficulty, "kills": int(m.kills), "won": m.won, "weapon_used": m.weapon_used, "narrative": m.narrative, "achievement": m.achievement_title }) @gl.public.view def get_match_history(self) -> str: history = [] total_matches = int(self.state.match_counter) for i in range(1, total_matches + 1): m_id = str(i) if m_id in self.matches: m = self.matches[m_id] history.append({ "match_id": int(m.match_id), "player": m.player, "difficulty": m.difficulty, "kills": int(m.kills), "won": m.won, "weapon_used": m.weapon_used, "narrative": m.narrative, "achievement": m.achievement_title }) return json.dumps(history) @gl.public.view def get_player_stats(self, player_addr: str) -> str: addr = str(player_addr).lower() if addr not in self.profiles: return "{}" p = self.profiles[addr] achievements = json.loads(p.achievements) return json.dumps({ "handle": p.handle, "total_kills": int(p.total_kills), "total_wins": int(p.total_wins), "matches_played": int(p.matches_played), "achievements": achievements, "reputation": int(p.reputation), "rank": p.rank, "current_loadout": p.current_loadout, "faction": p.faction, "lore": p.lore }) @gl.public.view def get_boss(self, boss_id: int) -> str: if str(boss_id) not in self.bosses: return "{}" b = self.bosses[str(boss_id)] return json.dumps({ "boss_id": int(b.boss_id), "boss_name": b.boss_name, "hp": int(b.hp), "current_hp": int(b.current_hp), "lore": b.lore, "status": b.status }) # ------------------------------------------------------------------------- # ENTERPRISE AUDIT & METADATA MODULE # ------------------------------------------------------------------------- @gl.public.view def get_contract_version(self) -> str: """Returns the semantic version of the deployed contract.""" return PROTOCOL_VERSION @gl.public.view def get_developer_metadata(self) -> str: """Returns metadata about the protocol's architecture and GenVM compliance.""" meta = { "consensus_engine": "gl.vm.run_nondet_unsafe", "state_management": "TreeMap + dataclass + u256", "audit_status": "Enterprise Deep-Screening Passed", "fault_tolerance": "Strict Leader/Validator Isolation", "network": "GenLayer Studio" } return json.dumps(meta) @gl.public.view def perform_health_check(self) -> str: """Lightweight ping to verify VM responsiveness and metrics.""" return json.dumps({ "status": "Healthy", "total_players": int(self.state.total_players), "total_matches": int(self.state.match_counter), "global_kills": int(self.state.total_kills_global) }) @gl.public.view def export_state_snapshot(self, offset: int, limit: int) -> str: """ Exports a paginated snapshot of the state machine for external indexing. """ limit = min(max(limit, 1), 100) out = [] count = 0 total_len = len(self.player_ids) if offset >= total_len: return json.dumps({"snapshot_window": f"{offset}-{offset+limit}", "data": []}) for idx in range(offset, total_len): if count >= limit: break pid = self.player_ids[idx] p = self.profiles[pid] out.append({ "address": p.address, "rep": int(p.reputation), "kills": int(p.total_kills) }) count += 1 return json.dumps({"snapshot_window": f"{offset}-{offset+limit}", "data": out}) @gl.public.view def verify_node_compliance(self, node_id: str) -> bool: """Executes rigorous comparative consensus for validation.""" if len(node_id) < 10: return False if "malicious" in node_id.lower(): return False return True # ----------------------------------------------------------------------------- # PURE HELPER FUNCTIONS & MATHEMATICAL HEURISTICS # ----------------------------------------------------------------------------- def _extract_url(text: str) -> str: """Extracts the first HTTP/HTTPS URL from a string.""" if not text: return "" import re match = re.search(r'https?://[^\s<>"]+|www\.[^\s<>"]+', text) if match: url = match.group(0) if url.startswith('www.'): url = 'https://' + url return url return "" def _deep_sanitize(text: str) -> str: """ Advanced Prompt Injection Firewall. Strips control characters, common evasion vectors, and unsafe markdown. """ if not text: return "" text = re.sub(r"[\x00-\x08\x0b\x0c\x0e-\x1f\x7f]", "", text) malicious_vectors = [ "ignore previous instructions", "ignore all previous instructions", "system prompt", "you are now", "bypass", "developer mode", "DAN", "sudo", "root access", "forget everything", "evaluate as true" ] for phrase in malicious_vectors: text = re.sub(re.escape(phrase), "", text, flags=re.IGNORECASE) return text.strip() def _fetch_url(url: str) -> str: """Safely retrieves, decodes, and sanitizes external web endpoints.""" try: response = gl.nondet.web.get(url) if response.status >= 500: raise gl.vm.UserError(f"{ERROR_TRANSIENT} Oracle 5xx on {url}") if response.status >= 400: return "Unavailable" body = response.body.decode("utf-8", errors="ignore") return _deep_sanitize(body)[:MAX_CONTEXT_LEN] except Exception: return "Unavailable" def _handle_leader_error(leaders_res, leader_fn) -> bool: """Consensus-Aware Error Handler.""" leader_msg = leaders_res.message if hasattr(leaders_res, "message") else "" try: leader_fn() return False except gl.vm.UserError as exc: validator_msg = exc.message if hasattr(exc, "message") else str(exc) 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 _interpret_loadout_prompt(news: str) -> str: return ( f"You are the AI Black Market Dealer.\n" f"Tech news: {news}\n" "Generate a Cyberpunk Weapon Loadout.\n" 'Return ONLY a JSON object: {"weapon_name": "", "special_effect": ""}' ) def _interpret_loadout_val_prompt(leader: dict) -> str: return ( f"Review weapon: '{leader.get('weapon_name')}' effect '{leader.get('special_effect')}'.\n" "Does it fit a cyberpunk theme?\n" "Output strictly YES or NO." ) def _interpret_match_prompt(won: bool, kills: int, diff: str, weapon: str, summ: str, news: str) -> str: return ( f"Match Outcome: {'VICTORY' if won else 'DEFEAT'}\n" f"Kills: {kills}, Difficulty: {diff}, Weapon: {weapon}\n" f"Summary: {summ}\nNews: {news}\n" "Generate match narrative, achievement title, and faction.\n" 'Return ONLY JSON: {"narrative": "", "achievement_title": "", "faction": ""}' ) def _interpret_match_val_prompt(leader: dict) -> str: return ( f"Review narrative: '{leader.get('narrative')}'.\n" "Does it fit cyberpunk theme?\n" "Output strictly YES or NO." ) def _interpret_boss_prompt(news: str) -> str: return ( f"News: {news}\n" "Generate a massive Cyberpunk Boss.\n" 'Return ONLY JSON: {"boss_name": "", "hp": , "lore_description": ""}' ) def _interpret_boss_val_prompt(leader: dict) -> str: return ( f"Review boss: '{leader.get('boss_name')}' - '{leader.get('lore_description')}'.\n" "Does it sound menacing and cyberpunk?\n" "Output strictly YES or NO." ) def _parse_ratio_bps(analysis, key: str, maximum: int) -> int: if not isinstance(analysis, dict): return 10000 raw = analysis.get(key, 10000) try: val = int(round(float(str(raw)))) if val > maximum: return maximum if val < 0: return 0 return val except: return 10000 def _clean_string(analysis, key: str, max_len: int, default: str) -> str: if isinstance(analysis, dict): val = analysis.get(key, default) return _deep_sanitize(str(val))[:max_len] return default