# { "Depends": "py-genlayer:1jb45aa8ynh2a9c9xn3b7qqh8sm5q93hwfp7jqmwsfhh8jpz09h6" } from dataclasses import dataclass import typing from genlayer import * MAX_MESSAGE_CHARS = 280 MAX_FINAL_MESSAGE_CHARS = 420 MAX_RUN_NOTE_CHARS = 280 MAX_REASON_CHARS = 96 MAX_TITLE_CHARS = 48 MAX_PLAYER_NAME_CHARS = 24 MAX_WEEK_ID_CHARS = 32 MIN_REASONABLE_RUN_SECONDS = 30.0 MAX_REASONABLE_RUN_SECONDS = 7200.0 @allow_storage @dataclass class GuardianOathRecord: player: str oathMessage: str accepted: bool score: u32 category: str reason: str title: str @allow_storage @dataclass class FinalDecisionRecord: player: str finalMessage: str dashUnlocked: bool doubleJumpUnlocked: bool scrapHoundDefeated: bool reactorTitanDefeated: bool coreRestored: bool accepted: bool score: u32 category: str reason: str title: str @allow_storage @dataclass class WeeklyRunRecord: player: str weekId: str playerName: str completionTimeSeconds: float dashUnlocked: bool doubleJumpUnlocked: bool scrapHoundDefeated: bool reactorTitanDefeated: bool coreRestored: bool runNote: str accepted: bool score: u32 category: str reason: str rankEligible: bool title: str class MochiProtocolAdjudicator(gl.Contract): schema_version: str guardian_oaths: TreeMap[str, GuardianOathRecord] final_decisions: TreeMap[str, FinalDecisionRecord] weekly_runs: DynArray[WeeklyRunRecord] player_best_weekly_runs: TreeMap[str, WeeklyRunRecord] def __init__(self): self.schema_version = "MochiProtocolAdjudicator/v1" @gl.public.write def submit_guardian_oath(self, oath_message: str) -> typing.Any: clean_message = self._clean_text(oath_message, MAX_MESSAGE_CHARS) if len(clean_message) == 0: return self._reject("guardian_oath", "Oath message is empty.", "Silent Oath") if len(str(oath_message)) > MAX_MESSAGE_CHARS: return self._reject("guardian_oath", "Oath message is too long.", "Oversized Oath") prompt = ( "You are the Intelligent Layer adjudicator for the game Mochi Protocol.\n" "Mochi is a tiny robot cat guardian. The validators stopped agreeing, " "the Core fractured, and Mochi must restore consensus inside the Broken Core.\n\n" "Judge this Guardian Oath. Accept only if it is short, clear, not hostile, " "not unrelated nonsense, and expresses intent to restore, repair, reconnect, " "protect, or help. It should relate to the Intelligent Layer, Core, consensus, " "validators, Mochi, or the broken network.\n\n" "Return only JSON with exactly these fields:\n" '{"accepted": boolean, "score": integer 0-100, "category": "guardian_oath", ' '"reason": "short reason", "title": "short oath title"}\n\n' "Guardian oath:\n" f"{clean_message}" ) result = self._judge_with_llm(prompt, "guardian_oath", False) if result["accepted"]: player = self._sender_key() self.guardian_oaths[player] = GuardianOathRecord( player=player, oathMessage=clean_message, accepted=result["accepted"], score=u32(result["score"]), category=result["category"], reason=result["reason"], title=result["title"], ) return result @gl.public.write def submit_final_decision( self, progress_data: typing.Any, final_message: str ) -> typing.Any: progress = self._normalize_progress(progress_data) clean_message = self._clean_text(final_message, MAX_FINAL_MESSAGE_CHARS) if len(clean_message) == 0: return self._reject( "final_decision", "Final message is empty.", "Silent Ending" ) if len(str(final_message)) > MAX_FINAL_MESSAGE_CHARS: return self._reject( "final_decision", "Final message is too long.", "Oversized Ending" ) if not self._progress_complete(progress): return self._reject( "final_decision", "Required restoration progress is incomplete.", "Core Not Ready", ) prompt = ( "You are the Intelligent Layer adjudicator for the game Mochi Protocol.\n" "The player claims the final restoration is complete after restoring Dash, " "restoring Double Jump, defeating Scrap Hound, defeating Reactor Titan, " "and restoring the Core.\n\n" "Judge whether the final message fits a valid story ending. It should express " "restoration, reconnection, consensus, completion, or protection. Reject hostile, " "unrelated, or nonsensical submissions.\n\n" "Return only JSON with exactly these fields:\n" '{"accepted": boolean, "score": integer 0-100, "category": "final_decision", ' '"reason": "short reason", "title": "short ending title"}\n\n' "Progress data: dashUnlocked=true, doubleJumpUnlocked=true, " "scrapHoundDefeated=true, reactorTitanDefeated=true, coreRestored=true\n" "Final message:\n" f"{clean_message}" ) result = self._judge_with_llm(prompt, "final_decision", False) if result["accepted"]: player = self._sender_key() self.final_decisions[player] = FinalDecisionRecord( player=player, finalMessage=clean_message, dashUnlocked=True, doubleJumpUnlocked=True, scrapHoundDefeated=True, reactorTitanDefeated=True, coreRestored=True, accepted=result["accepted"], score=u32(result["score"]), category=result["category"], reason=result["reason"], title=result["title"], ) return result @gl.public.write def submit_weekly_run(self, run_data: typing.Any) -> typing.Any: run = self._normalize_run_data(run_data) precheck_failure = self._weekly_run_precheck(run) if precheck_failure != "": return self._reject_weekly(precheck_failure, "Run Rejected") prompt = ( "You are the Intelligent Layer adjudicator for Mochi Protocol weekly speedruns.\n" "The game is a sci-fi robot cat Metroidvania. A valid demo completion restores " "Dash, restores Double Jump, defeats Scrap Hound, defeats Reactor Titan, and " "restores the Core.\n\n" "Judge whether this weekly run submission looks valid and rank-eligible. " "The run note should describe the run, restoration, completion, route, bosses, " "abilities, or Core repair in a relevant way. Reject unrelated, unsafe, hostile, " "or nonsensical notes. Rank eligibility should be true only when the submission " "is accepted and suitable for display on the leaderboard.\n\n" "Return only JSON with exactly these fields:\n" '{"accepted": boolean, "score": integer 0-100, "category": "weekly_speedrun", ' '"reason": "short reason", "rankEligible": boolean, ' '"title": "short display title"}\n\n' f"Run data: completionTimeSeconds={run['completionTimeSeconds']}, " f"weekId={run['weekId']}, playerName={run['playerName']}, " "dashUnlocked=true, doubleJumpUnlocked=true, scrapHoundDefeated=true, " "reactorTitanDefeated=true, coreRestored=true\n" f"Run note:\n{run['runNote']}" ) result = self._judge_with_llm(prompt, "weekly_speedrun", True) result["rankEligible"] = bool(result["accepted"] and result["rankEligible"]) if result["accepted"]: player = self._sender_key() record = WeeklyRunRecord( player=player, weekId=run["weekId"], playerName=run["playerName"], completionTimeSeconds=run["completionTimeSeconds"], dashUnlocked=True, doubleJumpUnlocked=True, scrapHoundDefeated=True, reactorTitanDefeated=True, coreRestored=True, runNote=run["runNote"], accepted=result["accepted"], score=u32(result["score"]), category=result["category"], reason=result["reason"], rankEligible=result["rankEligible"], title=result["title"], ) self.weekly_runs.append(record) if record.rankEligible: best_key = self._best_run_key(player, record.weekId) if best_key not in self.player_best_weekly_runs: self.player_best_weekly_runs[best_key] = record else: current_best = self.player_best_weekly_runs[best_key] if self._is_better_run(record, current_best): self.player_best_weekly_runs[best_key] = record return result @gl.public.view def get_weekly_leaderboard(self, week_id: str) -> typing.Any: clean_week_id = self._clean_text(week_id, MAX_WEEK_ID_CHARS) entries = [] for record in self.weekly_runs: if ( record.weekId == clean_week_id and record.accepted and record.rankEligible ): entries.append(self._weekly_record_to_dict(record)) entries.sort(key=lambda entry: (entry["completionTimeSeconds"], -entry["score"])) return entries @gl.public.view def get_player_best_run(self, player: str, week_id: str) -> typing.Any: clean_player = self._clean_text(player, 128) clean_week_id = self._clean_text(week_id, MAX_WEEK_ID_CHARS) best_key = self._best_run_key(clean_player, clean_week_id) if best_key not in self.player_best_weekly_runs: return {} return self._weekly_record_to_dict(self.player_best_weekly_runs[best_key]) @gl.public.view def get_player_record(self, player: str) -> typing.Any: clean_player = self._clean_text(player, 128) oath = {} final = {} best_runs = [] if clean_player in self.guardian_oaths: oath = self._oath_record_to_dict(self.guardian_oaths[clean_player]) if clean_player in self.final_decisions: final = self._final_record_to_dict(self.final_decisions[clean_player]) for record in self.weekly_runs: if record.player == clean_player and record.accepted and record.rankEligible: best_key = self._best_run_key(clean_player, record.weekId) if ( best_key in self.player_best_weekly_runs and self.player_best_weekly_runs[best_key].completionTimeSeconds == record.completionTimeSeconds and self.player_best_weekly_runs[best_key].score == record.score ): best_runs.append(self._weekly_record_to_dict(record)) best_runs.sort(key=lambda entry: entry["weekId"]) return { "player": clean_player, "guardianOath": oath, "finalDecision": final, "bestWeeklyRuns": best_runs, } def _judge_with_llm( self, prompt: str, category: str, include_rank_eligible: bool ) -> typing.Any: def leader_fn() -> typing.Any: raw = gl.nondet.exec_prompt(prompt, response_format="json") return self._normalize_llm_result(raw, category, include_rank_eligible) def validator_fn(leader_result: typing.Any) -> bool: if not isinstance(leader_result, gl.vm.Return): return False return self._valid_result_shape( leader_result.calldata, category, include_rank_eligible ) return gl.vm.run_nondet_unsafe(leader_fn, validator_fn) def _normalize_llm_result( self, raw: typing.Any, category: str, include_rank_eligible: bool ) -> typing.Any: if not isinstance(raw, dict): if include_rank_eligible: return self._reject_weekly("Invalid LLM output.", "Invalid Judgment") return self._reject(category, "Invalid LLM output.", "Invalid Judgment") accepted = bool(raw.get("accepted", False)) score = self._clamp_score(raw.get("score", 0)) reason = self._clean_text(str(raw.get("reason", "")), MAX_REASON_CHARS) title = self._clean_text(str(raw.get("title", "")), MAX_TITLE_CHARS) if reason == "": reason = "No clear reason provided." accepted = False if title == "": title = "Untitled Judgment" result = { "accepted": accepted, "score": score, "category": category, "reason": reason, "title": title, } if include_rank_eligible: result["rankEligible"] = bool(raw.get("rankEligible", False)) return result def _valid_result_shape( self, result: typing.Any, category: str, include_rank_eligible: bool ) -> bool: if not isinstance(result, dict): return False if result.get("category") != category: return False if not isinstance(result.get("accepted"), bool): return False if not isinstance(result.get("score"), int): return False if result["score"] < 0 or result["score"] > 100: return False if not isinstance(result.get("reason"), str): return False if not isinstance(result.get("title"), str): return False if len(result["reason"]) == 0 or len(result["reason"]) > MAX_REASON_CHARS: return False if len(result["title"]) == 0 or len(result["title"]) > MAX_TITLE_CHARS: return False if include_rank_eligible and not isinstance(result.get("rankEligible"), bool): return False return True def _reject(self, category: str, reason: str, title: str) -> typing.Any: return { "accepted": False, "score": 0, "category": category, "reason": self._clean_text(reason, MAX_REASON_CHARS), "title": self._clean_text(title, MAX_TITLE_CHARS), } def _reject_weekly(self, reason: str, title: str) -> typing.Any: return { "accepted": False, "score": 0, "category": "weekly_speedrun", "reason": self._clean_text(reason, MAX_REASON_CHARS), "rankEligible": False, "title": self._clean_text(title, MAX_TITLE_CHARS), } def _sender_key(self) -> str: sender = str(gl.message.sender_address) if sender == "" or sender == "None": return "studio_sender" return sender def _normalize_progress(self, progress_data: typing.Any) -> typing.Any: if not isinstance(progress_data, dict): return { "dashUnlocked": False, "doubleJumpUnlocked": False, "scrapHoundDefeated": False, "reactorTitanDefeated": False, "coreRestored": False, } return { "dashUnlocked": bool(progress_data.get("dashUnlocked", False)), "doubleJumpUnlocked": bool(progress_data.get("doubleJumpUnlocked", False)), "scrapHoundDefeated": bool(progress_data.get("scrapHoundDefeated", False)), "reactorTitanDefeated": bool( progress_data.get("reactorTitanDefeated", False) ), "coreRestored": bool(progress_data.get("coreRestored", False)), } def _progress_complete(self, progress: typing.Any) -> bool: return ( progress["dashUnlocked"] and progress["doubleJumpUnlocked"] and progress["scrapHoundDefeated"] and progress["reactorTitanDefeated"] and progress["coreRestored"] ) def _normalize_run_data(self, run_data: typing.Any) -> typing.Any: data = run_data if isinstance(run_data, dict) else {} return { "completionTimeSeconds": self._safe_float( data.get("completionTimeSeconds", 0) ), "dashUnlocked": bool(data.get("dashUnlocked", False)), "doubleJumpUnlocked": bool(data.get("doubleJumpUnlocked", False)), "scrapHoundDefeated": bool(data.get("scrapHoundDefeated", False)), "reactorTitanDefeated": bool(data.get("reactorTitanDefeated", False)), "coreRestored": bool(data.get("coreRestored", False)), "weekId": self._clean_text(str(data.get("weekId", "")), MAX_WEEK_ID_CHARS), "playerName": self._clean_text( str(data.get("playerName", "")), MAX_PLAYER_NAME_CHARS ), "runNote": self._clean_text(str(data.get("runNote", "")), MAX_RUN_NOTE_CHARS), } def _weekly_run_precheck(self, run: typing.Any) -> str: if run["completionTimeSeconds"] <= 0: return "Completion time must be greater than zero." if run["completionTimeSeconds"] < MIN_REASONABLE_RUN_SECONDS: return "Completion time is below the demo range." if run["completionTimeSeconds"] > MAX_REASONABLE_RUN_SECONDS: return "Completion time is above the demo range." if ( not run["dashUnlocked"] or not run["doubleJumpUnlocked"] or not run["scrapHoundDefeated"] or not run["reactorTitanDefeated"] or not run["coreRestored"] ): return "Required restoration progress is incomplete." if run["weekId"] == "": return "Week ID is required." if not self._safe_identifier(run["weekId"], MAX_WEEK_ID_CHARS): return "Week ID is not display-safe." if run["playerName"] == "": return "Player name is required." if not self._safe_player_name(run["playerName"]): return "Player name is not display-safe." if run["runNote"] == "": return "Run note is required." return "" def _safe_float(self, value: typing.Any) -> float: try: return float(value) except Exception: return 0.0 def _clamp_score(self, value: typing.Any) -> int: try: score = int(value) except Exception: score = 0 if score < 0: return 0 if score > 100: return 100 return score def _clean_text(self, value: str, max_chars: int) -> str: text = str(value).replace("\n", " ").replace("\r", " ").strip() while " " in text: text = text.replace(" ", " ") if len(text) > max_chars: return text[:max_chars].strip() return text def _safe_identifier(self, text: str, max_chars: int) -> bool: if len(text) == 0 or len(text) > max_chars: return False for char in text: if not ( ("a" <= char <= "z") or ("A" <= char <= "Z") or ("0" <= char <= "9") or char == "-" or char == "_" ): return False return True def _safe_player_name(self, text: str) -> bool: if len(text) == 0 or len(text) > MAX_PLAYER_NAME_CHARS: return False for char in text: if not ( ("a" <= char <= "z") or ("A" <= char <= "Z") or ("0" <= char <= "9") or char == "-" or char == "_" or char == "." or char == " " ): return False return True def _best_run_key(self, player: str, week_id: str) -> str: return f"{player}|{week_id}" def _is_better_run(self, candidate: WeeklyRunRecord, current: WeeklyRunRecord) -> bool: if candidate.completionTimeSeconds < current.completionTimeSeconds: return True if candidate.completionTimeSeconds == current.completionTimeSeconds: return candidate.score > current.score return False def _oath_record_to_dict(self, record: GuardianOathRecord) -> typing.Any: return { "player": record.player, "oathMessage": record.oathMessage, "accepted": record.accepted, "score": int(record.score), "category": record.category, "reason": record.reason, "title": record.title, } def _final_record_to_dict(self, record: FinalDecisionRecord) -> typing.Any: return { "player": record.player, "finalMessage": record.finalMessage, "progressData": { "dashUnlocked": record.dashUnlocked, "doubleJumpUnlocked": record.doubleJumpUnlocked, "scrapHoundDefeated": record.scrapHoundDefeated, "reactorTitanDefeated": record.reactorTitanDefeated, "coreRestored": record.coreRestored, }, "accepted": record.accepted, "score": int(record.score), "category": record.category, "reason": record.reason, "title": record.title, } def _weekly_record_to_dict(self, record: WeeklyRunRecord) -> typing.Any: return { "player": record.player, "weekId": record.weekId, "playerName": record.playerName, "completionTimeSeconds": record.completionTimeSeconds, "progressData": { "dashUnlocked": record.dashUnlocked, "doubleJumpUnlocked": record.doubleJumpUnlocked, "scrapHoundDefeated": record.scrapHoundDefeated, "reactorTitanDefeated": record.reactorTitanDefeated, "coreRestored": record.coreRestored, }, "runNote": record.runNote, "accepted": record.accepted, "score": int(record.score), "category": record.category, "reason": record.reason, "rankEligible": record.rankEligible, "title": record.title, }