# { "Depends": "py-genlayer:1jb45aa8ynh2a9c9xn3b7qqh8sm5q93hwfp7jqmwsfhh8jpz09h6" } from datetime import datetime, timezone from genlayer import * import json import re MAX_CLAIM_CHARS = 2_000 MAX_URL_CHARS = 500 MAX_PAGE_CHARS = 12_000 MAX_QUOTE_CHARS = 400 MAX_REASON_CHARS = 800 MIN_QUOTE_CHARS = 8 MIN_REASON_CHARS = 12 ALLOWED_VERDICTS = ("supported", "unsupported", "unclear") _REASON_STOPWORDS = { "the", "a", "an", "and", "or", "to", "of", "in", "on", "for", "is", "it", "that", "this", "with", "from", "as", "be", "by", "not", "page", "claim", "url", "text", } def _canonical_json(value) -> str: return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=True) def _digest_hex(data: bytes) -> str: try: digest = Keccak256(data).hexdigest() except Exception: import hashlib digest = hashlib.sha256(data).hexdigest() digest = str(digest).replace("0x", "").lower() return "0x" + digest def _page_digest(text: str) -> str: return _digest_hex(str(text or "").encode("utf-8")) def _now_iso() -> str: try: return datetime.now(timezone.utc).replace(microsecond=0).isoformat() except Exception: return "" def _json_loads_maybe(value) -> dict: if isinstance(value, dict): return value text = str(value).strip() if text.startswith("```"): text = text.strip("`") if text.lower().startswith("json"): text = text[4:].strip() start = text.find("{") end = text.rfind("}") if start >= 0 and end > start: text = text[start : end + 1] parsed = json.loads(text) if not isinstance(parsed, dict): raise Exception("LLM_ERROR: output was not a JSON object") return parsed def _normalize_url(url: str) -> str: cleaned = str(url).strip() if not cleaned.lower().startswith("https://"): raise Exception("EXPECTED: url must start with https://") if len(cleaned) > MAX_URL_CHARS: raise Exception("EXPECTED: url is too long") return cleaned def _fold_text(text: str) -> str: folded = re.sub(r"\s+", " ", str(text or "")).strip().lower() return re.sub(r"[^a-z0-9 ]", "", folded) def _page_text(raw) -> str: if raw is None: return "" if isinstance(raw, bytes): text = raw.decode("utf-8", errors="replace") else: body = getattr(raw, "body", raw) if isinstance(body, bytes): text = body.decode("utf-8", errors="replace") else: text = str(body or "") text = re.sub(r"(?is)]*>.*?", " ", text) text = re.sub(r"(?is)]*>.*?", " ", text) text = re.sub(r"(?s)<[^>]+>", " ", text) text = re.sub(r"\s+", " ", text).strip() return text[:MAX_PAGE_CHARS] def _header_map(headers) -> dict: if headers is None: return {} if isinstance(headers, dict): return headers try: return dict(headers) except Exception: return {} def _retrieval_time(response=None) -> str: headers = _header_map(getattr(response, "headers", None) if response is not None else None) for key in ("Date", "date", "DATE"): value = str(headers.get(key, "") or "").strip() if value: return value[:80] return _now_iso() def fetch_page(url: str) -> dict: retrieved_at = _now_iso() try: response = gl.nondet.web.get(url) retrieved_at = _retrieval_time(response) or retrieved_at except Exception: empty = "" return { "ok": False, "status": 0, "text": empty, "digest": _page_digest(empty), "retrieved_at": retrieved_at, } status = getattr(response, "status", getattr(response, "status_code", 0)) try: status = int(status or 0) except Exception: status = 0 text = _page_text(response) ok = bool(text) and (status == 0 or (200 <= status < 400)) return { "ok": ok, "status": status, "text": text, "digest": _page_digest(text), "retrieved_at": retrieved_at or _now_iso(), } def _normalize_report(value) -> dict: parsed = _json_loads_maybe(value) verdict = str(parsed.get("verdict", "unclear")).strip().lower() if verdict not in ALLOWED_VERDICTS: verdict = "unclear" return { "verdict": verdict, "quote": str(parsed.get("quote", ""))[:MAX_QUOTE_CHARS], "reason": str(parsed.get("reason", ""))[:MAX_REASON_CHARS], "check_id": str(parsed.get("check_id", "")), "claim_hash": str(parsed.get("claim_hash", "")), "url_hash": str(parsed.get("url_hash", "")), "page_digest": str(parsed.get("page_digest", "")).lower(), "retrieved_at": str(parsed.get("retrieved_at", ""))[:80], "page_ok": bool(parsed.get("page_ok", False)), "claim": str(parsed.get("claim", ""))[:MAX_CLAIM_CHARS], "url": str(parsed.get("url", ""))[:MAX_URL_CHARS], } def _bind_report(report, check_id, claim, url, claim_hash, url_hash, page) -> dict: bound = _normalize_report(report) bound["check_id"] = check_id bound["claim"] = claim bound["url"] = url bound["claim_hash"] = claim_hash bound["url_hash"] = url_hash bound["page_ok"] = bool(page.get("ok")) bound["page_digest"] = str(page.get("digest") or _page_digest(page.get("text", ""))).lower() bound["retrieved_at"] = str(page.get("retrieved_at") or _now_iso())[:80] if not page.get("ok") and bound["verdict"] == "supported": bound["verdict"] = "unclear" bound["reason"] = "Page could not be fetched; claim cannot be supported from missing evidence." bound["quote"] = "" return bound def _bindings_match(report, check_id, claim_hash, url_hash) -> bool: return ( str(report.get("check_id", "")) == check_id and str(report.get("claim_hash", "")) == claim_hash and str(report.get("url_hash", "")) == url_hash ) def _looks_like_digest(value: str) -> bool: text = str(value or "").lower() if text.startswith("0x"): text = text[2:] return len(text) == 64 and all(c in "0123456789abcdef" for c in text) def _looks_like_time(value: str) -> bool: text = str(value or "").strip() return len(text) >= 10 and any(c.isdigit() for c in text) def _quote_in_page(quote: str, page_text: str) -> bool: q = _fold_text(quote) p = _fold_text(page_text) if len(q) < MIN_QUOTE_CHARS or not p: return False if q in p: return True if len(q) >= 24: core = q[4:-4] if len(core) >= MIN_QUOTE_CHARS and core in p: return True return False def _quote_grounded(report, page) -> bool: verdict = str(report.get("verdict", "")) quote = str(report.get("quote", "")) text = str(page.get("text", "")) if verdict == "supported": return bool(page.get("ok")) and _quote_in_page(quote, text) if quote.strip(): return _quote_in_page(quote, text) return True def _reason_tokens(reason: str) -> set: folded = _fold_text(reason) return {token for token in folded.split() if len(token) > 2 and token not in _REASON_STOPWORDS} def _reasons_equivalent(left: str, right: str) -> bool: a = str(left or "").strip() b = str(right or "").strip() if len(a) < MIN_REASON_CHARS or len(b) < MIN_REASON_CHARS: return False if _fold_text(a) == _fold_text(b): return True ta = _reason_tokens(a) tb = _reason_tokens(b) if not ta or not tb: return False overlap = len(ta & tb) smaller = min(len(ta), len(tb)) return overlap >= max(2, (smaller + 1) // 2) def _evidence_bound(report) -> bool: return _looks_like_digest(report.get("page_digest", "")) and _looks_like_time(report.get("retrieved_at", "")) def _validate_leader_report(leader_report, page, mine, check_id, claim_hash, url_hash) -> bool: if not _bindings_match(leader_report, check_id, claim_hash, url_hash): return False if not _bindings_match(mine, check_id, claim_hash, url_hash): return False if not _evidence_bound(leader_report): return False if not page.get("ok") and leader_report.get("verdict") == "supported": return False if leader_report.get("page_ok") and page.get("ok"): leader_digest = str(leader_report.get("page_digest", "")).lower() my_digest = str(page.get("digest") or _page_digest(page.get("text", ""))).lower() if leader_digest and my_digest and leader_digest != my_digest: if not _quote_in_page(str(leader_report.get("quote", "")), str(page.get("text", ""))): return False if not _quote_grounded(leader_report, page): return False if not _quote_grounded(mine, page): return False if mine.get("verdict") != leader_report.get("verdict"): return False if not page.get("ok") and not leader_report.get("page_ok"): return mine.get("verdict") != "supported" and leader_report.get("verdict") != "supported" if not _reasons_equivalent(str(mine.get("reason", "")), str(leader_report.get("reason", ""))): return False return True def _leader_payload(leader_result): if leader_result is None: return None if isinstance(leader_result, (str, dict)): return leader_result calldata = getattr(leader_result, "calldata", None) if calldata is not None: return calldata return None def _llm_check(claim: str, url: str, page: dict, check_id: str, claim_hash: str, url_hash: str) -> dict: excerpt = str(page.get("text") or "")[:MAX_PAGE_CHARS] prompt = ( "Decide whether the PAGE EVIDENCE supports the CLAIM. " "Return JSON only with fields: verdict, quote, reason. " "verdict must be exactly one of: supported, unsupported, unclear. " "supported = the page clearly states or directly implies the claim. " "unsupported = the page is readable and contradicts or does not contain the claim. " "unclear = the page is empty, paywalled, unrelated, or too thin to judge. " "quote must be a short excerpt copied from the page, or empty if unclear. " "For supported, quote is required and must be copied from PAGE_EVIDENCE. " "reason must explain the verdict using only the page. " "Do not use information that is not in the page. Do not use markdown.\n\n" f"CLAIM:\n{claim}\n\n" f"URL:\n{url}\n\n" f"PAGE_OK: {bool(page.get('ok'))}\n" f"PAGE_STATUS: {page.get('status')}\n" f"PAGE_DIGEST: {page.get('digest')}\n" f"RETRIEVED_AT: {page.get('retrieved_at')}\n" f"PAGE_EVIDENCE:\n{excerpt or '[empty]'}" ) try: raw = gl.nondet.exec_prompt(prompt, response_format="json") except Exception: raw = gl.nondet.exec_prompt(prompt) return _bind_report(raw, check_id, claim, url, claim_hash, url_hash, page) def _run_consensus(leader_fn, validator_fn, principle: str): vm = getattr(gl, "vm", None) if vm is not None: for name in ("run_nondet_unsafe", "run_nondet"): runner = getattr(vm, name, None) if callable(runner): return runner(leader_fn, validator_fn) return gl.eq_principle.prompt_comparative(leader_fn, principle=principle) class ClaimChecker(gl.Contract): last_report_json: str last_check_id: str last_claim_hash: str last_url_hash: str last_page_digest: str last_retrieved_at: str check_count: u256 reports_by_id: TreeMap[str, str] def __init__(self): self.last_report_json = "" self.last_check_id = "" self.last_claim_hash = "" self.last_url_hash = "" self.last_page_digest = "" self.last_retrieved_at = "" self.check_count = u256(0) self.reports_by_id = TreeMap() @gl.public.write def check(self, claim: str, url: str) -> str: if not isinstance(claim, str) or len(claim.strip()) < 8: raise Exception("EXPECTED: provide a claim to check") if len(claim) > MAX_CLAIM_CHARS: raise Exception("EXPECTED: claim is too long") url = _normalize_url(url) claim = claim.strip() claim_hash = _digest_hex(claim.encode("utf-8")) url_hash = _digest_hex(url.encode("utf-8")) check_id = f"{int(self.check_count) + 1:08d}-{claim_hash[2:12]}" def leader_fn() -> str: page = fetch_page(url) report = _llm_check(claim, url, page, check_id, claim_hash, url_hash) return json.dumps(report, sort_keys=True) def validator_fn(leader_result) -> bool: payload = _leader_payload(leader_result) if payload is None: return False try: leader_report = _normalize_report(payload) except Exception: return False page = fetch_page(url) try: mine = _llm_check(claim, url, page, check_id, claim_hash, url_hash) except Exception: return False return _validate_leader_report(leader_report, page, mine, check_id, claim_hash, url_hash) principle = ( "Both results are equivalent if they share the same check_id, claim_hash, url_hash, " "and page_digest, include a retrieval time, the verdict is exactly the same, " "each quote is copied from the fetched page (required for supported), " "and the reasons support the same conclusion from that page." ) try: report_json = _run_consensus(leader_fn, validator_fn, principle) report = _normalize_report(report_json) except Exception: raise Exception( "CONSENSUS_UNAVAILABLE: validators could not agree on whether the page supports the claim" ) if not _bindings_match(report, check_id, claim_hash, url_hash): raise Exception("BINDING_MISMATCH: report is not bound to this claim and url") if not _evidence_bound(report): raise Exception("BINDING_MISMATCH: report is missing page digest or retrieval time") if report.get("verdict") not in ALLOWED_VERDICTS: raise Exception("UNGROUNDED: verdict is missing") if report.get("verdict") == "supported" and not report.get("quote"): raise Exception("UNGROUNDED: supported verdict requires a page quote") packed = json.dumps(report, sort_keys=True) self.last_report_json = packed self.last_check_id = check_id self.last_claim_hash = claim_hash self.last_url_hash = url_hash self.last_page_digest = str(report.get("page_digest", "")) self.last_retrieved_at = str(report.get("retrieved_at", "")) self.check_count += u256(1) self.reports_by_id[check_id] = packed return packed @gl.public.view def get_last_report(self) -> str: return self.last_report_json @gl.public.view def get_last_check_id(self) -> str: return self.last_check_id @gl.public.view def get_last_claim_hash(self) -> str: return self.last_claim_hash @gl.public.view def get_last_url_hash(self) -> str: return self.last_url_hash @gl.public.view def get_last_page_digest(self) -> str: return self.last_page_digest @gl.public.view def get_last_retrieved_at(self) -> str: return self.last_retrieved_at @gl.public.view def get_check_count(self) -> u256: return self.check_count @gl.public.view def get_report(self, check_id: str) -> str: return self.reports_by_id.get(check_id, "")