# { "Depends": "py-genlayer:1jb45aa8ynh2a9c9xn3b7qqh8sm5q93hwfp7jqmwsfhh8jpz09h6" } import json import typing from dataclasses import dataclass from genlayer import * # --- Error classification prefixes ------------------------------------------- ERROR_EXPECTED = "[EXPECTED]" # Business logic — deterministic, must match exactly ERROR_LLM = "[LLM_ERROR]" # LLM misbehavior — validators disagree, force rotation # --- Game constants ---------------------------------------------------------- NUM_QUESTIONS = 5 CORRECT_THRESHOLD = 50 # score >= this counts as a "correct" answer SCORE_TOLERANCE = 35 # max allowed leader/validator score divergence TOPICS = "GenLayer, Web3, and Crypto" # --- Storage models ---------------------------------------------------------- @allow_storage @dataclass class Round: # All fields are certified storage primitives (no nested dynamic collections), # so the storage layout / schema generates on every GenVM version. Questions and # the player roster are kept in flat top-level TreeMaps keyed by round id. id: u256 status: str # "open" | "finished" jackpot: u256 # accumulated entry fees, atto-scale winner: Address joined_count: u256 finished_count: u256 best_score: u256 @allow_storage @dataclass class PlayerScore: player: Address total_score: u256 answers_given: u256 finished: bool # --- LLM-output helpers (defensive parsing) ---------------------------------- def _parse_questions(resp: typing.Any) -> list: """Normalize an LLM question-generation response into 5 question dicts.""" if not isinstance(resp, dict): raise gl.vm.UserError(f"{ERROR_LLM} Question response not a dict: {type(resp)}") raw = resp.get("questions") if raw is None: for alt in ("items", "list", "data", "result"): if alt in resp: raw = resp[alt] break if not isinstance(raw, list): raise gl.vm.UserError(f"{ERROR_LLM} Missing 'questions' list. Keys: {list(resp.keys())}") out = [] for i, item in enumerate(raw[:NUM_QUESTIONS]): if isinstance(item, dict): text = item.get("question") or item.get("text") or item.get("q") or "" topic = item.get("topic") or item.get("category") or TOPICS else: text = str(item) topic = TOPICS text = str(text).strip() if len(text) < 10: raise gl.vm.UserError(f"{ERROR_LLM} Question {i} too short: {text!r}") out.append({"id": i, "question": text, "topic": str(topic).strip()}) if len(out) != NUM_QUESTIONS: raise gl.vm.UserError(f"{ERROR_LLM} Expected {NUM_QUESTIONS} questions, got {len(out)}") return out def _parse_score(resp: typing.Any) -> int: """Extract a 0-100 integer score from an LLM grading response.""" if not isinstance(resp, dict): raise gl.vm.UserError(f"{ERROR_LLM} Grade response not a dict: {type(resp)}") raw = resp.get("score") if raw is None: for alt in ("rating", "points", "value", "grade", "result"): if alt in resp: raw = resp[alt] break if raw is None: raise gl.vm.UserError(f"{ERROR_LLM} Missing 'score'. Keys: {list(resp.keys())}") try: score = int(round(float(str(raw).strip()))) except (ValueError, TypeError): raise gl.vm.UserError(f"{ERROR_LLM} Non-numeric score: {raw!r}") return max(0, min(100, score)) def _handle_leader_error(leaders_res: gl.vm.Result, leader_fn) -> bool: """Decide agreement when the leader returned an error instead of a value.""" leader_msg = getattr(leaders_res, "message", "") try: leader_fn() return False # Leader errored but validator succeeded — disagree except gl.vm.UserError as e: validator_msg = getattr(e, "message", str(e)) # Deterministic business errors must match exactly. if validator_msg.startswith(ERROR_EXPECTED): return validator_msg == leader_msg # LLM or unknown errors: disagree to force consensus rotation. return False except Exception: return False class TriviaBattle(gl.Contract): owner: Address round_count: u256 # total rounds ever created current_round_id: u256 # 0 when no round is open rounds: TreeMap[u256, Round] scores: TreeMap[str, PlayerScore] # key: ":" answered: TreeMap[str, bool] # key: "::" questions: TreeMap[str, str] # key: ":" -> question JSON round_player_at: TreeMap[str, Address] # key: ":" -> player address def __init__(self): self.owner = gl.message.sender_address self.round_count = u256(0) self.current_round_id = u256(0) # --- Key helpers --------------------------------------------------------- def _player_key(self, round_id: int, player: Address) -> str: return f"{round_id}:{player.as_hex}" def _answer_key(self, round_id: int, player: Address, question_id: int) -> str: return f"{round_id}:{player.as_hex}:{question_id}" def _question_key(self, round_id: int, question_id: int) -> str: return f"{round_id}:{question_id}" def _player_at_key(self, round_id: int, index: int) -> str: return f"{round_id}:{index}" # --- 1. Start a round ---------------------------------------------------- @gl.public.write def start_round(self) -> dict: """Generate 5 AI trivia questions and open a new round.""" if int(self.current_round_id) != 0: raise gl.vm.UserError(f"{ERROR_EXPECTED} A round is already open") questions = self._generate_questions() round_id = u256(int(self.round_count) + 1) self.round_count = round_id self.current_round_id = round_id rnd = self.rounds.get_or_insert_default(round_id) rnd.id = round_id rnd.status = "open" rnd.jackpot = u256(0) rnd.joined_count = u256(0) rnd.finished_count = u256(0) rnd.best_score = u256(0) for q in questions: self.questions[self._question_key(int(round_id), q["id"])] = json.dumps(q, sort_keys=True) return {"round_id": int(round_id), "questions": questions} def _generate_questions(self) -> list: prompt = f"""You are a quizmaster. Generate exactly {NUM_QUESTIONS} factual trivia questions about {TOPICS}. Each question must have one objectively correct answer and cover a different subtopic. Do not include the answers. Respond ONLY with JSON in this exact shape: {{ "questions": [ {{"question": "", "topic": ""}} ] }}""" def leader_fn() -> dict: resp = gl.nondet.exec_prompt(prompt, response_format="json") return {"questions": _parse_questions(resp)} def validator_fn(leaders_res: gl.vm.Result) -> bool: # Open-ended generation has no single correct output, so the validator # performs a non-comparative quality check: the leader's batch must be # well-formed (exactly 5 distinct, non-trivial questions) rather than a # second candidate to diff against. if not isinstance(leaders_res, gl.vm.Return): return _handle_leader_error(leaders_res, leader_fn) qs = leaders_res.calldata.get("questions") if not isinstance(qs, list) or len(qs) != NUM_QUESTIONS: return False seen = set() for q in qs: if not isinstance(q, dict): return False text = str(q.get("question", "")).strip().lower() if len(text) < 10 or len(text) > 500: return False if text in seen: return False seen.add(text) return True result = gl.vm.run_nondet_unsafe(leader_fn, validator_fn) return result["questions"] # --- 2. Join a round (payable) ------------------------------------------ @gl.public.write.payable def join_game(self) -> None: """Pay the entry fee to join the open round; the fee feeds the jackpot.""" round_id = int(self.current_round_id) if round_id == 0: raise gl.vm.UserError(f"{ERROR_EXPECTED} No open round to join") rnd = self.rounds[u256(round_id)] if rnd.status != "open": raise gl.vm.UserError(f"{ERROR_EXPECTED} Round is not open") player = gl.message.sender_address key = self._player_key(round_id, player) if key in self.scores: raise gl.vm.UserError(f"{ERROR_EXPECTED} Already joined this round") self.scores[key] = PlayerScore( player=player, total_score=u256(0), answers_given=u256(0), finished=False, ) idx = int(rnd.joined_count) self.round_player_at[self._player_at_key(round_id, idx)] = player rnd.jackpot = u256(int(rnd.jackpot) + int(gl.message.value)) rnd.joined_count = u256(idx + 1) # --- 3 & 4. Answer a question, AI grades it ------------------------------ @gl.public.write def answer_question(self, question_id: int, answer: str) -> dict: """Submit an answer; AI validators grade it 0-100 and add it to the score.""" round_id = int(self.current_round_id) if round_id == 0: raise gl.vm.UserError(f"{ERROR_EXPECTED} No open round") rnd = self.rounds[u256(round_id)] if rnd.status != "open": raise gl.vm.UserError(f"{ERROR_EXPECTED} Round is not open") if question_id < 0 or question_id >= NUM_QUESTIONS: raise gl.vm.UserError(f"{ERROR_EXPECTED} Invalid question id") player = gl.message.sender_address pkey = self._player_key(round_id, player) if pkey not in self.scores: raise gl.vm.UserError(f"{ERROR_EXPECTED} Join the game first") ps = self.scores[pkey] if ps.finished: raise gl.vm.UserError(f"{ERROR_EXPECTED} Already finished this round") akey = self._answer_key(round_id, player, question_id) if self.answered.get(akey, False): raise gl.vm.UserError(f"{ERROR_EXPECTED} Question already answered") question = json.loads(self.questions[self._question_key(round_id, question_id)]) score = self._grade_answer(question["question"], answer) self.answered[akey] = True ps.total_score = u256(int(ps.total_score) + score) ps.answers_given = u256(int(ps.answers_given) + 1) return { "question_id": question_id, "score": score, "total_score": int(ps.total_score), "answers_given": int(ps.answers_given), } def _grade_answer(self, question: str, answer: str) -> int: prompt = f"""You are grading a trivia answer about {TOPICS}. Question: {question} Player's answer: {answer} Score the answer from 0 to 100 based purely on factual correctness: - 0 means completely wrong, empty, or irrelevant. - 100 means fully correct. - Partial credit for partially correct answers. Respond ONLY with JSON: {{"score": , "reasoning": ""}}""" def leader_fn() -> dict: resp = gl.nondet.exec_prompt(prompt, response_format="json") return {"score": _parse_score(resp), "reasoning": str(resp.get("reasoning", ""))} def validator_fn(leaders_res: gl.vm.Result) -> bool: # Grading decides money, so the validator independently re-grades and # must agree on the substantive outcome (correct vs incorrect bucket) # within a numeric tolerance. if not isinstance(leaders_res, gl.vm.Return): return _handle_leader_error(leaders_res, leader_fn) leader_score = int(leaders_res.calldata["score"]) validator_score = int(leader_fn()["score"]) if (leader_score >= CORRECT_THRESHOLD) != (validator_score >= CORRECT_THRESHOLD): return False if abs(leader_score - validator_score) > SCORE_TOLERANCE: return False return True result = gl.vm.run_nondet_unsafe(leader_fn, validator_fn) return int(result["score"]) # --- 5. Finish the game -------------------------------------------------- @gl.public.write def finish_game(self) -> dict: """Finalize the caller's score once all questions are answered.""" round_id = int(self.current_round_id) if round_id == 0: raise gl.vm.UserError(f"{ERROR_EXPECTED} No open round") rnd = self.rounds[u256(round_id)] if rnd.status != "open": raise gl.vm.UserError(f"{ERROR_EXPECTED} Round is not open") player = gl.message.sender_address pkey = self._player_key(round_id, player) if pkey not in self.scores: raise gl.vm.UserError(f"{ERROR_EXPECTED} Join the game first") ps = self.scores[pkey] if ps.finished: raise gl.vm.UserError(f"{ERROR_EXPECTED} Already finished") if int(ps.answers_given) < NUM_QUESTIONS: raise gl.vm.UserError(f"{ERROR_EXPECTED} Answer all questions first") ps.finished = True rnd.finished_count = u256(int(rnd.finished_count) + 1) # First finisher is the leader by default, so the winner is always a real # player (never the zero address) even if the top score is 0. Later finishers # must strictly beat the best score, so the earliest top scorer keeps ties. final_score = int(ps.total_score) if int(rnd.finished_count) == 1 or final_score > int(rnd.best_score): rnd.best_score = u256(final_score) rnd.winner = player return { "round_id": round_id, "final_score": final_score, "leading": rnd.winner == player, } # --- 6. Settle: pay the jackpot to the winner ---------------------------- @gl.public.write def end_round(self) -> dict: """Close the round once everyone has finished and pay the jackpot.""" round_id = int(self.current_round_id) if round_id == 0: raise gl.vm.UserError(f"{ERROR_EXPECTED} No open round") rnd = self.rounds[u256(round_id)] if rnd.status != "open": raise gl.vm.UserError(f"{ERROR_EXPECTED} Round is not open") if int(rnd.joined_count) == 0: raise gl.vm.UserError(f"{ERROR_EXPECTED} No players joined") # Anyone may settle once every joined player has finished. To stop a round # being locked forever by a player who never finishes, the owner can # force-settle — but only once at least one player has finished, so there is # always a real winner to pay. all_finished = int(rnd.finished_count) >= int(rnd.joined_count) if not all_finished: if gl.message.sender_address != self.owner: raise gl.vm.UserError(f"{ERROR_EXPECTED} Not all players have finished") if int(rnd.finished_count) == 0: raise gl.vm.UserError(f"{ERROR_EXPECTED} No players have finished yet") rnd.status = "finished" self.current_round_id = u256(0) winner = rnd.winner jackpot = int(rnd.jackpot) if jackpot > 0: gl.get_contract_at(winner).emit_transfer(value=u256(jackpot), on="finalized") return { "round_id": round_id, "winner": winner.as_hex, "jackpot": jackpot, "best_score": int(rnd.best_score), } # --- Views --------------------------------------------------------------- @gl.public.view def get_current_round_id(self) -> int: return int(self.current_round_id) @gl.public.view def get_round(self, round_id: int) -> dict: if u256(round_id) not in self.rounds: raise gl.vm.UserError(f"{ERROR_EXPECTED} Unknown round") rnd = self.rounds[u256(round_id)] return { "id": int(rnd.id), "status": rnd.status, "jackpot": int(rnd.jackpot), "winner": rnd.winner.as_hex, "joined_count": int(rnd.joined_count), "finished_count": int(rnd.finished_count), "best_score": int(rnd.best_score), "num_questions": NUM_QUESTIONS, } @gl.public.view def get_questions(self, round_id: int) -> list: if u256(round_id) not in self.rounds: raise gl.vm.UserError(f"{ERROR_EXPECTED} Unknown round") return [ json.loads(self.questions[self._question_key(round_id, i)]) for i in range(NUM_QUESTIONS) ] @gl.public.view def get_leaderboard(self, round_id: int) -> list: """All players in a round, ranked by score (highest first).""" if u256(round_id) not in self.rounds: return [] rnd = self.rounds[u256(round_id)] count = int(rnd.joined_count) # `board` is a plain in-method Python list (not storage), built from the # flat round_player_at roster. board = [] for i in range(count): addr = self.round_player_at[self._player_at_key(round_id, i)] ps = self.scores[self._player_key(round_id, addr)] board.append({ "player": addr.as_hex, "total_score": int(ps.total_score), "answers_given": int(ps.answers_given), "finished": ps.finished, }) # Manual selection sort by total_score, descending (no lambda / no key fn). n = len(board) for i in range(n): best = i for j in range(i + 1, n): if board[j]["total_score"] > board[best]["total_score"]: best = j if best != i: board[i], board[best] = board[best], board[i] return board @gl.public.view def get_player_score(self, round_id: int, player: str) -> dict: key = f"{round_id}:{Address(player).as_hex}" if key not in self.scores: raise gl.vm.UserError(f"{ERROR_EXPECTED} Player not in round") ps = self.scores[key] return { "player": ps.player.as_hex, "total_score": int(ps.total_score), "answers_given": int(ps.answers_given), "finished": ps.finished, }