# { "Depends": "py-genlayer:1jb45aa8ynh2a9c9xn3b7qqh8sm5q93hwfp7jqmwsfhh8jpz09h6" } """ Qwik — on-chain verification registry for the Qwik app. The web app (`src/lib/verify.functions.ts`) currently runs a three-persona LLM panel plus an arbiter on the server and keeps the outcome in localStorage. That makes the server the sole authority: nothing about the verdict is reproducible or contestable by anyone else. This contract moves the part that actually needs consensus on-chain: frontend/backend owns : UI, input capture, rendering, caching, analytics this contract owns : evidence input, the validator panel, the arbiter rule, the settled verdict, the appeal path, the public history external sources own : raw facts (web pages, GitHub API) — never trusted blindly; every GenLayer validator re-fetches them Consensus works on two levels: 1. Inside one nondet block a panel of three personas (the same personas the app uses) each returns a structured verdict, and a *deterministic* arbiter folds them into a final status / confidence / agreement. Determinism here matters: it means two honest nodes that get similar persona verdicts land on the exact same aggregate. 2. Every GenLayer validator re-runs that whole panel independently and only agrees if the final status matches exactly, the same evidence source was used, and the confidence is within tolerance. The leader's answer is never accepted on its own word. """ import json import re import urllib.parse from dataclasses import dataclass from genlayer import * # -------------------------------------------------------------------------- # Error classification # # Validators need to know whether an error is reproducible. Deterministic # errors must match exactly; transient ones may differ in wording; LLM # misbehaviour must always disagree so consensus rotates to another leader. # -------------------------------------------------------------------------- ERROR_EXPECTED = "[EXPECTED]" # business logic — deterministic ERROR_EXTERNAL = "[EXTERNAL]" # external 4xx / bad payload — deterministic ERROR_TRANSIENT = "[TRANSIENT]" # network / 5xx — non-deterministic ERROR_LLM = "[LLM_ERROR]" # model returned garbage — force rotation # -------------------------------------------------------------------------- # Domain constants # -------------------------------------------------------------------------- STATUS_TRUE = "True" STATUS_FALSE = "False" STATUS_MIXED = "Mixed" STATUS_INSUFFICIENT = "Insufficient Evidence" ALL_STATUSES = (STATUS_TRUE, STATUS_FALSE, STATUS_MIXED, STATUS_INSUFFICIENT) # The six verification modules shipped by the app, in route order. # (key, display title, verification task, evidence strategy) MODULES = ( ( "claim", "Claim Verification", "Verify the truthfulness of the following public claim.", "none", ), ( "document", "Document Verification", "Verify the authenticity indicators (issuer, structure, plausibility) of the described document.", "none", ), ( "website", "Website & Domain Trust", "Assess trust signals for this website/domain (age, SSL, reputation, identity consistency).", "web", ), ( "wallet", "Wallet & Token", "Assess on-chain risk & reputation for this wallet/token address.", "none", ), ( "company", "Company Verification", "Verify the legitimacy of this company/organization (registration, presence).", "none", ), ( "repository", "Repository Audit", "Assess the trust and health of this open-source repository (license, activity, risks).", "github", ), ) MODULE_KEYS = tuple(m[0] for m in MODULES) # The panel personas mirror the ones in src/lib/verify.functions.ts. Models are # deliberately not pinned here — on GenLayer the model diversity comes from the # validator set, not from the contract. PANEL = ( ("qwik-alpha", "skeptical fact-checker who requires strong evidence"), ("qwik-beta", "analytical researcher who weighs sources by authority"), ("qwik-gamma", "neutral evaluator focused on consistency of claims"), ) MIN_SUBJECT_LEN = 3 MAX_SUBJECT_LEN = 4000 MAX_EVIDENCE_CHARS = 6000 MAX_EVIDENCE_ITEMS = 8 MAX_FINDINGS = 6 MAX_PAGE_SIZE = 50 MAX_REVISIONS = 4 # 1 initial run + 3 appeals # How far two nodes' confidence may drift before they are considered to # disagree. LLM confidence is soft, so a wide band is correct here — the # hard gate is `status`, which must match exactly. CONFIDENCE_TOLERANCE = 30 # -------------------------------------------------------------------------- # Storage # -------------------------------------------------------------------------- @allow_storage @dataclass class Verification: """One settled verification. New fields go at the END only — storage layout is positional and inserting in the middle breaks a deployed contract.""" requester: Address module: str subject: str status: str confidence: u256 # 0..100 agreement: u256 # 0..100, share of the panel behind `status` headline: str key_findings: str # JSON array of strings reports: str # JSON array of per-persona verdicts evidence: str # JSON array of evidence objects evidence_source: str # url actually fetched, or "" when LLM-only created_at: str settled_at: str revision: u256 # 1 on first settle, +1 per successful appeal previous_status: str # status before the last appeal, "" if never appealed class QwikConsensus(gl.Contract): owner: Address paused: bool verifications: DynArray[Verification] # id lists — a verification's id is its index in `verifications` by_requester: TreeMap[str, DynArray[u256]] by_module: TreeMap[str, DynArray[u256]] # O(1) dashboard counters, kept in step with the collections above status_counts: TreeMap[str, u256] module_counts: TreeMap[str, u256] # keccak(module|subject) -> id + 1 (0 means "never verified") cache_index: TreeMap[str, u256] def __init__(self) -> None: self.owner = gl.message.sender_address self.paused = False # ---------------------------------------------------------------- writes @gl.public.write def verify(self, module: str, subject: str) -> u256: """Run the validator panel over `subject` and settle the result on-chain. Returns the new verification id. Callers that want to reuse an earlier answer should check `find_cached` first — this method always runs a fresh panel.""" self._require_active() module_key = _require_module(module) clean_subject = _require_subject(subject) result = _run_consensus(module_key, clean_subject) now = _now() vid = u256(len(self.verifications)) self.verifications.append( Verification( requester=gl.message.sender_address, module=module_key, subject=clean_subject, status=result["status"], confidence=u256(result["confidence"]), agreement=u256(result["agreement"]), headline=result["headline"], key_findings=json.dumps(result["key_findings"]), reports=json.dumps(result["reports"]), evidence=json.dumps(result["evidence"]), evidence_source=result["evidence_source"], created_at=now, settled_at=now, revision=u256(1), previous_status="", ) ) self.by_requester.get_or_insert_default( gl.message.sender_address.as_hex ).append(vid) self.by_module.get_or_insert_default(module_key).append(vid) self._bump(self.status_counts, result["status"]) self._bump(self.module_counts, module_key) self.cache_index[_cache_key(module_key, clean_subject)] = u256(vid + 1) return vid @gl.public.write def appeal(self, verification_id: u256) -> str: """Re-run the panel on an existing verification and settle the outcome. This is the contract-level appeal: anyone may pay to have the panel look again, and the record keeps the status it held before the re-run so a flip is visible rather than silent. (GenLayer's own `genlayer appeal` re-runs consensus on a single transaction; this re-runs the *decision*.) """ self._require_active() idx = self._require_id(verification_id) record = self.verifications[idx] if int(record.revision) >= MAX_REVISIONS: raise gl.vm.UserError( f"{ERROR_EXPECTED} Appeal limit reached for verification {idx}" ) result = _run_consensus(str(record.module), str(record.subject)) old_status = str(record.status) if old_status != result["status"]: self._unbump(self.status_counts, old_status) self._bump(self.status_counts, result["status"]) record.previous_status = old_status record.status = result["status"] record.confidence = u256(result["confidence"]) record.agreement = u256(result["agreement"]) record.headline = result["headline"] record.key_findings = json.dumps(result["key_findings"]) record.reports = json.dumps(result["reports"]) record.evidence = json.dumps(result["evidence"]) record.evidence_source = result["evidence_source"] record.settled_at = _now() record.revision = u256(int(record.revision) + 1) return result["status"] @gl.public.write def set_paused(self, paused: bool) -> None: """Circuit breaker — stops new verifications and appeals.""" self._require_owner() self.paused = paused @gl.public.write def transfer_ownership(self, new_owner: str) -> None: self._require_owner() try: parsed = Address(new_owner) except Exception: raise gl.vm.UserError(f"{ERROR_EXPECTED} Invalid owner address") self.owner = parsed # ----------------------------------------------------------------- views @gl.public.view def get_verification(self, verification_id: u256) -> dict: idx = self._require_id(verification_id) return _serialize(self.verifications[idx], idx) @gl.public.view def get_recent(self, limit: u256) -> list: """Newest verifications first — backs the dashboard history list.""" count = len(self.verifications) take = _clamp_page(limit) out = [] i = count - 1 while i >= 0 and len(out) < take: out.append(_serialize(self.verifications[i], i)) i -= 1 return out @gl.public.view def get_by_requester(self, requester: str, limit: u256) -> list: """Per-account history — the on-chain replacement for localStorage.""" try: key = Address(requester).as_hex except Exception: raise gl.vm.UserError(f"{ERROR_EXPECTED} Invalid requester address") ids = self.by_requester.get(key, None) if ids is None: return [] take = _clamp_page(limit) out = [] i = len(ids) - 1 while i >= 0 and len(out) < take: idx = int(ids[i]) out.append(_serialize(self.verifications[idx], idx)) i -= 1 return out @gl.public.view def get_by_module(self, module: str, limit: u256) -> list: module_key = _require_module(module) ids = self.by_module.get(module_key, None) if ids is None: return [] take = _clamp_page(limit) out = [] i = len(ids) - 1 while i >= 0 and len(out) < take: idx = int(ids[i]) out.append(_serialize(self.verifications[idx], idx)) i -= 1 return out @gl.public.view def find_cached(self, module: str, subject: str) -> int: """Id of the latest verification for this exact subject, or -1. Lets the frontend show a settled answer instantly instead of paying for a fresh panel run on a repeated question.""" module_key = _require_module(module) clean_subject = _require_subject(subject) found = self.cache_index.get(_cache_key(module_key, clean_subject), u256(0)) return int(found) - 1 @gl.public.view def get_stats(self) -> dict: """Aggregate counters for the dashboard — O(1), no scanning.""" return { "total": len(self.verifications), "by_status": { s: int(self.status_counts.get(s, u256(0))) for s in ALL_STATUSES }, "by_module": { k: int(self.module_counts.get(k, u256(0))) for k in MODULE_KEYS }, } @gl.public.view def get_modules(self) -> list: """Module catalogue — backs /products and the landing page module ring.""" return [ {"key": key, "title": title, "task": task, "evidence": evidence} for key, title, task, evidence in MODULES ] @gl.public.view def get_panel(self) -> list: """Persona lineup — backs the /consensus validator explanation.""" return [{"id": pid, "persona": persona} for pid, persona in PANEL] @gl.public.view def get_config(self) -> dict: return { "owner": self.owner.as_hex, "paused": self.paused, "statuses": list(ALL_STATUSES), "max_revisions": MAX_REVISIONS, "confidence_tolerance": CONFIDENCE_TOLERANCE, "min_subject_len": MIN_SUBJECT_LEN, "max_subject_len": MAX_SUBJECT_LEN, } # ------------------------------------------------------------- internals def _require_owner(self) -> None: if gl.message.sender_address != self.owner: raise gl.vm.UserError(f"{ERROR_EXPECTED} Only owner") def _require_active(self) -> None: if self.paused: raise gl.vm.UserError(f"{ERROR_EXPECTED} Contract is paused") def _require_id(self, verification_id: u256) -> int: idx = int(verification_id) if idx < 0 or idx >= len(self.verifications): raise gl.vm.UserError(f"{ERROR_EXPECTED} Unknown verification {idx}") return idx def _bump(self, counter: TreeMap[str, u256], key: str) -> None: counter[key] = u256(int(counter.get(key, u256(0))) + 1) def _unbump(self, counter: TreeMap[str, u256], key: str) -> None: current = int(counter.get(key, u256(0))) counter[key] = u256(current - 1 if current > 0 else 0) # -------------------------------------------------------------------------- # Consensus engine # # Everything below is module level on purpose: the leader and validator # closures are serialized to run in sub-VMs, so they must not capture `self` # or any storage handle. # -------------------------------------------------------------------------- def _run_consensus(module_key: str, subject: str) -> dict: """Run the panel under a leader/validator equivalence principle. The validator does not inspect the leader's JSON shape — it re-runs the entire panel itself and compares the substantive outcome. A leader that invents a verdict cannot pass this, because an honest validator will have produced its own verdict from the same evidence.""" def leader_fn() -> dict: return _panel_run(module_key, subject) def validator_fn(leaders_res: gl.vm.Result) -> bool: if not isinstance(leaders_res, gl.vm.Return): return _handle_leader_error(leaders_res, module_key, subject) leader = leaders_res.calldata if not isinstance(leader, dict): return False mine = _panel_run(module_key, subject) # Hard gate: the decision itself must match. if leader.get("status") != mine["status"]: return False # Both nodes must have looked at the same place. A leader that quietly # fetched a different URL is not verifying the same thing. if leader.get("evidence_source") != mine["evidence_source"]: return False leader_confidence = leader.get("confidence") if not isinstance(leader_confidence, int): return False if leader_confidence < 0 or leader_confidence > 100: return False if abs(leader_confidence - mine["confidence"]) > CONFIDENCE_TOLERANCE: return False return True return gl.vm.run_nondet_unsafe(leader_fn, validator_fn) def _handle_leader_error(leaders_res, module_key: str, subject: str) -> bool: """Decide whether to agree with a leader that failed. Deterministic failures must reproduce exactly. Transient ones only need to be transient on both sides. Anything LLM-shaped or unrecognised is a disagreement, which rotates the leader instead of freezing bad state.""" leader_msg = getattr(leaders_res, "message", "") try: _panel_run(module_key, subject) return False # leader failed where we succeeded — disagree except gl.vm.UserError as e: validator_msg = getattr(e, "message", str(e)) 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 _panel_run(module_key: str, subject: str) -> dict: """One full panel pass: gather evidence, poll three personas, arbitrate.""" task = _module_task(module_key) evidence_kind = _module_evidence_kind(module_key) source, corpus = _gather_evidence(evidence_kind, subject) reports = [ _persona_verdict(pid, persona, task, subject, corpus) for pid, persona in PANEL ] return _arbitrate(reports, source) def _persona_verdict( persona_id: str, persona: str, task: str, subject: str, corpus: str ) -> dict: """Ask one persona for a structured verdict and normalize it hard. LLMs drift on key names, wrap JSON in prose, and return confidence as a string or a 0..1 fraction. All of that is repaired here so the arbiter sees a clean, comparable shape.""" evidence_block = ( f"\n\nFetched source material (authoritative, prefer it over recall):\n{corpus}" if corpus else "\n\nNo source material was fetched for this module." ) prompt = ( f"You are an independent AI validator ({persona}) for the Qwik " f"verification network. Ground your verdict in verifiable evidence. If there " f"is not enough real evidence, answer \"{STATUS_INSUFFICIENT}\" with low " f"confidence. Never invent URLs — omit the url field when unsure.\n\n" f"Verification task: {task}\n\n" f"Subject:\n{subject}" f"{evidence_block}\n\n" f"Reply with JSON only:\n" f'{{"status": one of {list(ALL_STATUSES)}, "confidence": integer 0-100, ' f'"summary": short sentence, "reasoning": 2-3 sentences, ' f'"evidence": [{{"title": str, "url": str (optional), "snippet": str, ' f'"relevance": integer 0-100}}]}}' ) raw = gl.nondet.exec_prompt(prompt, response_format="json") if not isinstance(raw, dict): raise gl.vm.UserError( f"{ERROR_LLM} Persona {persona_id} returned {type(raw).__name__}, expected object" ) return { "validator": persona_id, "persona": persona, "status": _parse_status(raw, persona_id), "confidence": _parse_confidence(raw, persona_id), "summary": _parse_text(raw, ("summary", "verdict", "conclusion"), 400), "reasoning": _parse_text(raw, ("reasoning", "rationale", "explanation"), 1200), "evidence": _parse_evidence(raw), } def _arbitrate(reports: list, source: str) -> dict: """Fold persona verdicts into one settled outcome — deterministically. The app used a fourth LLM call for this. A plain function is strictly better here: given the same three verdicts every node computes the same aggregate, so consensus only has to survive the panel's variance, not the arbiter's.""" if not reports: raise gl.vm.UserError(f"{ERROR_LLM} Panel produced no verdicts") total = len(reports) tally = {status: 0 for status in ALL_STATUSES} for report in reports: tally[report["status"]] += 1 # Ties resolve by ALL_STATUSES order, which is fixed — never by dict order. top_status = ALL_STATUSES[0] top_count = tally[top_status] for status in ALL_STATUSES[1:]: if tally[status] > top_count: top_status = status top_count = tally[status] tied = [s for s in ALL_STATUSES if tally[s] == top_count] agreement = (top_count * 100) // total if len(tied) > 1: # A split panel is a finding in itself. Reporting "Mixed" is honest; # picking a side would launder disagreement into a verdict. final_status = STATUS_MIXED agreeing = reports elif top_status == STATUS_INSUFFICIENT: final_status = STATUS_INSUFFICIENT agreeing = [r for r in reports if r["status"] == top_status] elif top_count * 2 <= total: # A plurality is not a majority. Compared against `total` rather than # the rounded `agreement` percentage, because 2 of 3 floors to 66. final_status = STATUS_MIXED agreeing = reports else: final_status = top_status agreeing = [r for r in reports if r["status"] == top_status] raw_confidence = sum(r["confidence"] for r in agreeing) // len(agreeing) # Disagreement costs confidence — a 2:1 panel should never read as certain. confidence = (raw_confidence * agreement) // 100 lead = agreeing[0] for report in agreeing[1:]: if report["confidence"] > lead["confidence"]: lead = report headline = lead["summary"] or f"{final_status} ({agreement}% panel agreement)" findings = [] for report in reports: line = report["summary"].strip() if line and line not in findings: findings.append(line) if len(findings) >= MAX_FINDINGS: break return { "status": final_status, "confidence": confidence, "agreement": agreement, "headline": headline[:280], "key_findings": findings, "evidence": _merge_evidence(reports), "reports": reports, "evidence_source": source, } def _merge_evidence(reports: list) -> list: """Union of every persona's evidence, strongest first, deduped.""" merged = [] seen = set() for report in reports: for item in report["evidence"]: key = (item["title"].lower(), item["url"]) if key in seen: continue seen.add(key) merged.append(item) merged.sort(key=lambda item: -item["relevance"]) return merged[:MAX_EVIDENCE_ITEMS] # -------------------------------------------------------------------------- # Evidence gathering # -------------------------------------------------------------------------- def _gather_evidence(kind: str, subject: str) -> tuple: """Return (source, corpus). `source` is compared across validators, so it must be derived from the subject alone — never from a live response.""" if kind == "web": return _fetch_website(subject) if kind == "github": return _fetch_repository(subject) return "", "" def _fetch_website(subject: str) -> tuple: url = _normalize_url(subject) try: text = gl.nondet.web.render(url, mode="text") except gl.vm.UserError: raise except Exception as e: raise gl.vm.UserError(f"{ERROR_TRANSIENT} Could not render {url}: {type(e).__name__}") if not isinstance(text, str) or not text.strip(): raise gl.vm.UserError(f"{ERROR_EXTERNAL} {url} returned no readable content") return url, text[:MAX_EVIDENCE_CHARS] def _fetch_repository(subject: str) -> tuple: api_url = _github_api_url(subject) try: res = gl.nondet.web.get( api_url, headers={ "Accept": "application/vnd.github+json", "User-Agent": "qwik-consensus", }, ) except gl.vm.UserError: raise except Exception as e: raise gl.vm.UserError( f"{ERROR_TRANSIENT} GitHub request failed: {type(e).__name__}" ) if res.status == 404: raise gl.vm.UserError(f"{ERROR_EXTERNAL} Repository not found: {api_url}") if 400 <= res.status < 500: raise gl.vm.UserError(f"{ERROR_EXTERNAL} GitHub returned {res.status}") if res.status >= 500: raise gl.vm.UserError(f"{ERROR_TRANSIENT} GitHub temporarily unavailable") if res.body is None: raise gl.vm.UserError(f"{ERROR_TRANSIENT} GitHub returned an empty body") try: data = json.loads(res.body.decode("utf-8")) except Exception: raise gl.vm.UserError(f"{ERROR_EXTERNAL} GitHub returned malformed JSON") if not isinstance(data, dict): raise gl.vm.UserError(f"{ERROR_EXTERNAL} GitHub returned an unexpected payload") license_info = data.get("license") license_id = "" if isinstance(license_info, dict): license_id = str(license_info.get("spdx_id") or license_info.get("key") or "") # Only stable-ish fields. Star and fork counts are bucketed because raw # counts move between the leader's fetch and a validator's fetch. facts = { "full_name": str(data.get("full_name") or ""), "description": str(data.get("description") or "")[:500], "license": license_id or "NONE", "archived": bool(data.get("archived")), "disabled": bool(data.get("disabled")), "is_fork": bool(data.get("fork")), "language": str(data.get("language") or ""), "default_branch": str(data.get("default_branch") or ""), "created_at": str(data.get("created_at") or ""), "last_push_month": str(data.get("pushed_at") or "")[:7], "stars_bucket": _bucket(data.get("stargazers_count")), "forks_bucket": _bucket(data.get("forks_count")), "open_issues_bucket": _bucket(data.get("open_issues_count")), } return api_url, json.dumps(facts, sort_keys=True) def _normalize_url(subject: str) -> str: candidate = subject.strip().split()[0] if subject.strip() else "" if not candidate: raise gl.vm.UserError(f"{ERROR_EXPECTED} No URL found in subject") if not candidate.lower().startswith(("http://", "https://")): candidate = "https://" + candidate parsed = urllib.parse.urlparse(candidate) if not parsed.netloc: raise gl.vm.UserError(f"{ERROR_EXPECTED} Invalid URL: {subject[:80]}") # Rebuild from parsed parts so two nodes normalize identically. Scheme and # host are case-insensitive per RFC 3986; the path is not. return urllib.parse.urlunparse( ( parsed.scheme.lower(), parsed.netloc.lower(), parsed.path or "/", "", parsed.query, "", ) ) def _github_api_url(subject: str) -> str: """Accept `owner/repo`, a github.com URL, or an api.github.com URL.""" text = subject.strip().split()[0] if subject.strip() else "" if not text: raise gl.vm.UserError(f"{ERROR_EXPECTED} No repository found in subject") if "github.com" in text: parsed = urllib.parse.urlparse( text if text.startswith("http") else "https://" + text ) parts = [p for p in parsed.path.split("/") if p] if parsed.netloc.lower() == "api.github.com" and parts[:1] == ["repos"]: parts = parts[1:] if len(parts) < 2: raise gl.vm.UserError(f"{ERROR_EXPECTED} Not a repository URL: {text[:80]}") owner, repo = parts[0], parts[1] else: parts = [p for p in text.split("/") if p] if len(parts) != 2: raise gl.vm.UserError( f"{ERROR_EXPECTED} Expected 'owner/repo' or a GitHub URL, got: {text[:80]}" ) owner, repo = parts[0], parts[1] if repo.endswith(".git"): repo = repo[:-4] if not _SLUG_RE.match(owner) or not _SLUG_RE.match(repo): raise gl.vm.UserError(f"{ERROR_EXPECTED} Invalid repository name: {text[:80]}") return f"https://api.github.com/repos/{owner}/{repo}" _SLUG_RE = re.compile(r"^[A-Za-z0-9._-]{1,100}$") def _bucket(value) -> str: """Collapse a churny count into a stable band.""" try: n = int(value) except (TypeError, ValueError): return "unknown" for limit, label in ((10, "0-10"), (100, "10-100"), (1000, "100-1k"), (10000, "1k-10k")): if n < limit: return label return "10k+" # -------------------------------------------------------------------------- # LLM response repair # -------------------------------------------------------------------------- def _parse_status(raw: dict, persona_id: str) -> str: value = raw.get("status") if value is None: for alt in ("verdict", "result", "label", "outcome"): if alt in raw: value = raw[alt] break if value is None: raise gl.vm.UserError( f"{ERROR_LLM} Persona {persona_id} omitted 'status'. Keys: {sorted(raw.keys())}" ) text = str(value).strip().lower() for status in ALL_STATUSES: if text == status.lower(): return status # Common paraphrases, checked before the insufficient-evidence catch-all so # "not enough evidence to call it true" doesn't get read as True. if "insufficient" in text or "unknown" in text or "unverifi" in text: return STATUS_INSUFFICIENT if "mixed" in text or "partial" in text or "misleading" in text: return STATUS_MIXED if text in ("true", "verified", "accurate", "correct", "yes", "legit", "legitimate"): return STATUS_TRUE if text in ("false", "fake", "inaccurate", "incorrect", "no", "scam", "fraudulent"): return STATUS_FALSE raise gl.vm.UserError( f"{ERROR_LLM} Persona {persona_id} returned unrecognised status: {str(value)[:60]}" ) def _parse_confidence(raw: dict, persona_id: str) -> int: value = raw.get("confidence") if value is None: for alt in ("score", "certainty", "confidence_score"): if alt in raw: value = raw[alt] break if value is None: return 50 # a missing confidence is a neutral one, not a failure try: number = float(str(value).strip().rstrip("%")) except (TypeError, ValueError): raise gl.vm.UserError( f"{ERROR_LLM} Persona {persona_id} returned non-numeric confidence: {str(value)[:40]}" ) # Models answer 0.82 as often as 82. if 0.0 < number <= 1.0: number = number * 100.0 return max(0, min(100, int(round(number)))) def _parse_text(raw: dict, keys: tuple, limit: int) -> str: for key in keys: value = raw.get(key) if isinstance(value, str) and value.strip(): return value.strip()[:limit] if isinstance(value, list) and value: return " ".join(str(v) for v in value).strip()[:limit] return "" def _parse_evidence(raw: dict) -> list: items = raw.get("evidence") if not isinstance(items, list): items = raw.get("sources") if isinstance(raw.get("sources"), list) else [] out = [] for item in items[:MAX_EVIDENCE_ITEMS]: if isinstance(item, str): out.append({"title": item[:200], "url": "", "snippet": "", "relevance": 50}) continue if not isinstance(item, dict): continue url = str(item.get("url") or item.get("link") or "") if url and not (url.startswith("http://") or url.startswith("https://")): url = "" # drop anything that isn't a real link rather than surface it relevance = item.get("relevance", item.get("score", 50)) try: relevance = max(0, min(100, int(round(float(str(relevance)))))) except (TypeError, ValueError): relevance = 50 out.append( { "title": str(item.get("title") or item.get("name") or "Evidence")[:200], "url": url[:500], "snippet": str(item.get("snippet") or item.get("summary") or "")[:500], "relevance": relevance, } ) return out # -------------------------------------------------------------------------- # Small helpers # -------------------------------------------------------------------------- def _module_task(module_key: str) -> str: for key, _title, task, _evidence in MODULES: if key == module_key: return task raise gl.vm.UserError(f"{ERROR_EXPECTED} Unknown module: {module_key}") def _module_evidence_kind(module_key: str) -> str: for key, _title, _task, evidence in MODULES: if key == module_key: return evidence raise gl.vm.UserError(f"{ERROR_EXPECTED} Unknown module: {module_key}") def _require_module(module: str) -> str: key = module.strip().lower() if key not in MODULE_KEYS: raise gl.vm.UserError( f"{ERROR_EXPECTED} Unknown module '{module[:40]}'. Expected one of {list(MODULE_KEYS)}" ) return key def _require_subject(subject: str) -> str: clean = " ".join(subject.split()) # collapse whitespace so the cache key is stable if len(clean) < MIN_SUBJECT_LEN: raise gl.vm.UserError( f"{ERROR_EXPECTED} Subject must be at least {MIN_SUBJECT_LEN} characters" ) if len(clean) > MAX_SUBJECT_LEN: raise gl.vm.UserError( f"{ERROR_EXPECTED} Subject must be at most {MAX_SUBJECT_LEN} characters" ) return clean def _cache_key(module_key: str, subject: str) -> str: hasher = Keccak256() hasher.update(f"{module_key}|{subject}".encode("utf-8")) return hasher.digest().hex() def _clamp_page(limit: u256) -> int: take = int(limit) if take <= 0: return MAX_PAGE_SIZE return min(take, MAX_PAGE_SIZE) def _now() -> str: """Transaction datetime — identical for every node executing this tx.""" return str(gl.message_raw["datetime"]) def _serialize(record: Verification, verification_id: int) -> dict: return { "id": verification_id, "requester": record.requester.as_hex, "module": str(record.module), "subject": str(record.subject), "status": str(record.status), "confidence": int(record.confidence), "agreement": int(record.agreement), "headline": str(record.headline), "keyFindings": json.loads(record.key_findings), "verdicts": json.loads(record.reports), "evidence": json.loads(record.evidence), "evidenceSource": str(record.evidence_source), "createdAt": str(record.created_at), "settledAt": str(record.settled_at), "revision": int(record.revision), "previousStatus": str(record.previous_status), }