# { "Depends": "py-genlayer:1jb45aa8ynh2a9c9xn3b7qqh8sm5q93hwfp7jqmwsfhh8jpz09h6" } from genlayer import * import hashlib import json import typing class ProofVerdict(typing.NamedTuple): milestone_hash: str criteria_hash: str trust_model_hash: str input_hash: str proof_hashes_json: str status: str risk: str confidence: u8 proof_count: u8 matched_criteria: u8 missing_criteria: u8 summary: str class MilestoneProofVerifier(gl.Contract): """Persistent registry for consensus-reviewed milestone completion proofs.""" proof_count: u64 latest_proof_id: str proof_ids: DynArray[str] proofs: TreeMap[str, str] def __init__(self): self.proof_count = u64(0) self.latest_proof_id = "" @gl.public.view def get_proof_count(self) -> u64: return self.proof_count @gl.public.view def get_latest_proof_id(self) -> str: return self.latest_proof_id @gl.public.view def get_proof(self, proof_id: str) -> str: return self.proofs.get(proof_id, "") @gl.public.view def list_proof_ids(self) -> str: return json.dumps([proof_id for proof_id in self.proof_ids], separators=(",", ":")) @gl.public.write def submit_proof( self, milestone: str, acceptance_criteria: str, source_trust_model: str, proof_urls: DynArray[str], ): self._store_review( "", "initial", "", milestone, acceptance_criteria, source_trust_model, proof_urls, ) @gl.public.write def request_re_review( self, parent_proof_id: str, challenge_reason: str, milestone: str, acceptance_criteria: str, source_trust_model: str, updated_proof_urls: DynArray[str], ): if len(parent_proof_id) == 0: raise Exception("parent proof id is required") if len(self.proofs.get(parent_proof_id, "")) == 0: raise Exception("parent proof does not exist") if len(challenge_reason) < 12: raise Exception("challenge reason is too short") self._store_review( parent_proof_id, "re_review", challenge_reason, milestone, acceptance_criteria, source_trust_model, updated_proof_urls, ) def _store_review( self, parent_proof_id: str, review_type: str, challenge_reason: str, milestone: str, acceptance_criteria: str, source_trust_model: str, proof_urls: DynArray[str], ): if len(milestone) < 12: raise Exception("milestone is too short") if len(acceptance_criteria) < 20: raise Exception("acceptance criteria are too short") if len(source_trust_model) < 20: raise Exception("source trust model is too short") if len(proof_urls) < 2: raise Exception("at least two proof URLs are required") if len(proof_urls) > 5: raise Exception("use five proof URLs or fewer") proof_list = [str(url) for url in proof_urls] def leader_fn(): return _verify_milestone(milestone, acceptance_criteria, source_trust_model, proof_list) def validator_fn(leader_result) -> bool: if not isinstance(leader_result, gl.vm.Return): return False try: proposed = _parse_verdict(leader_result.calldata) independent = _parse_verdict( _verify_milestone(milestone, acceptance_criteria, source_trust_model, proof_list) ) except Exception: return False if proposed.milestone_hash != independent.milestone_hash: return False if proposed.criteria_hash != independent.criteria_hash: return False if proposed.trust_model_hash != independent.trust_model_hash: return False if proposed.input_hash != independent.input_hash: return False if proposed.proof_hashes_json != independent.proof_hashes_json: return False if proposed.status != independent.status: return False if proposed.risk != independent.risk: return False if proposed.proof_count != independent.proof_count: return False if abs(int(proposed.confidence) - int(independent.confidence)) > 15: return False matched_delta = abs(int(proposed.matched_criteria) - int(independent.matched_criteria)) missing_delta = abs(int(proposed.missing_criteria) - int(independent.missing_criteria)) return matched_delta <= 1 and missing_delta <= 1 agreed_json = gl.vm.run_nondet_unsafe(leader_fn, validator_fn) verdict = json.loads(agreed_json) self.proof_count = u64(int(self.proof_count) + 1) proof_id = str(int(self.proof_count)) record = { "proof_id": proof_id, "parent_proof_id": parent_proof_id, "review_type": review_type, "challenge_reason": challenge_reason, "milestone": milestone, "acceptance_criteria": acceptance_criteria, "source_trust_model": source_trust_model, "proof_urls": proof_list, "submitted_by": str(gl.message.sender_address), "result": verdict, } self.proofs[proof_id] = json.dumps(record, sort_keys=True, separators=(",", ":")) self.proof_ids.append(proof_id) self.latest_proof_id = proof_id def _verify_milestone( milestone: str, acceptance_criteria: str, source_trust_model: str, proof_urls: typing.Sequence[str], ) -> str: proof_payloads = [] for url in proof_urls: page = gl.nondet.web.render(url, mode="text") proof_payloads.append( { "url": url, "text": page[:6000], } ) prompt = f""" You are verifying whether a submitted milestone has been completed. Milestone: {milestone} Acceptance criteria: {acceptance_criteria} Source trust model: {source_trust_model} Proof pages: {json.dumps(proof_payloads)} Return only minified JSON with exactly these keys: - status: one of "accepted", "rejected", "needs_review" - risk: one of "low", "medium", "high" - confidence: integer from 0 to 100 - matched_criteria: integer count of criteria clearly satisfied by the proof - missing_criteria: integer count of criteria not satisfied or not evidenced - summary: concise explanation under 450 characters Rules: - Accept only when the proof pages directly satisfy the milestone and criteria. - Apply the caller-selected source trust model when weighting proof pages. - Use "needs_review" when the proof is plausible but incomplete, ambiguous, or hard to verify. - Use "rejected" when proof pages are unrelated, inaccessible, or contradict the milestone. - Do not credit vague marketing language as completion evidence. - Prefer lower confidence when proof pages repeat each other or lack primary-source detail. """ raw = gl.nondet.exec_prompt(prompt) data = json.loads(raw) proof_count = len(proof_urls) proof_hashes = [_sha256(url) for url in proof_urls] milestone_hash = _sha256(milestone) criteria_hash = _sha256(acceptance_criteria) trust_model_hash = _sha256(source_trust_model) input_hash = _sha256( json.dumps( { "criteria_hash": criteria_hash, "milestone_hash": milestone_hash, "proof_hashes": proof_hashes, "trust_model_hash": trust_model_hash, }, sort_keys=True, separators=(",", ":"), ) ) normalized = { "confidence": max(0, min(100, int(data["confidence"]))), "criteria_hash": criteria_hash, "input_hash": input_hash, "matched_criteria": int(data["matched_criteria"]), "milestone_hash": milestone_hash, "missing_criteria": int(data["missing_criteria"]), "proof_count": proof_count, "proof_hashes": proof_hashes, "risk": str(data["risk"]).lower(), "status": str(data["status"]).lower(), "summary": str(data["summary"])[:450], "trust_model_hash": trust_model_hash, } return json.dumps(normalized, sort_keys=True, separators=(",", ":")) def _sha256(value: str) -> str: return hashlib.sha256(value.encode("utf-8")).hexdigest() def _parse_verdict(raw_json: str) -> ProofVerdict: data = json.loads(raw_json) milestone_hash = str(data["milestone_hash"]) criteria_hash = str(data["criteria_hash"]) trust_model_hash = str(data["trust_model_hash"]) input_hash = str(data["input_hash"]) proof_hashes = data["proof_hashes"] proof_hashes_json = json.dumps(proof_hashes, sort_keys=True, separators=(",", ":")) status = str(data["status"]).lower() risk = str(data["risk"]).lower() confidence = int(data["confidence"]) proof_count = int(data["proof_count"]) matched_criteria = int(data["matched_criteria"]) missing_criteria = int(data["missing_criteria"]) summary = str(data["summary"]) if ( len(milestone_hash) != 64 or len(criteria_hash) != 64 or len(trust_model_hash) != 64 or len(input_hash) != 64 ): raise Exception("invalid input hashes") if not isinstance(proof_hashes, list): raise Exception("invalid proof hashes") if len(proof_hashes) != proof_count: raise Exception("proof hash count mismatch") for proof_hash in proof_hashes: if len(str(proof_hash)) != 64: raise Exception("invalid proof hash") if status not in ("accepted", "rejected", "needs_review"): raise Exception("invalid status") if risk not in ("low", "medium", "high"): raise Exception("invalid risk") if confidence < 0 or confidence > 100: raise Exception("invalid confidence") if proof_count < 2 or proof_count > 5: raise Exception("invalid proof count") if matched_criteria < 0 or missing_criteria < 0: raise Exception("invalid criteria counts") if matched_criteria + missing_criteria < 1: raise Exception("criteria accounting is empty") if matched_criteria + missing_criteria > 12: raise Exception("criteria accounting exceeds limit") if len(summary) == 0 or len(summary) > 450: raise Exception("invalid summary") return ProofVerdict( milestone_hash=milestone_hash, criteria_hash=criteria_hash, trust_model_hash=trust_model_hash, input_hash=input_hash, proof_hashes_json=proof_hashes_json, status=status, risk=risk, confidence=u8(confidence), proof_count=u8(proof_count), matched_criteria=u8(matched_criteria), missing_criteria=u8(missing_criteria), summary=summary, )