# { "Depends": "py-genlayer:1jb45aa8ynh2a9c9xn3b7qqh8sm5q93hwfp7jqmwsfhh8jpz09h6" } """ IncidentCredit — SLA-pinned incident credit on GenLayer. A provider pins a public SLA page, binds a customer wallet, then locks GEN. A later public incident narrative is compared against that pinned SLA text. GEN pays the bound customer only if the incident qualifies; otherwise the credit stays funded for retry. After the claim window, unused GEN returns to the provider. """ import json from dataclasses import dataclass from genlayer import * import genlayer.gl.vm as glvm SLA_HOST_SUFFIXES = ( "cloudflare.com", "example.com", "example.org", ) INCIDENT_HOST_SUFFIXES = ( "cloudflarestatus.com", "example.com", "example.org", ) PAGE_LIMIT = 50 MIN_FUND_WEI = 10_000_000_000_000_000 # 0.01 GEN DEFAULT_CLAIM_WINDOW = 7 * 24 * 60 * 60 MAX_CLAIM_WINDOW = 90 * 24 * 60 * 60 SLA_TEXT_LIMIT = 8000 def _parse_llm_json(raw) -> dict: if isinstance(raw, dict): return raw s = str(raw).strip().replace("```json", "").replace("```", "").strip() start, end = s.find("{"), s.rfind("}") + 1 if start >= 0 and end > start: s = s[start:end] return json.loads(s) def _stable_json(data: dict) -> str: return json.dumps(data, sort_keys=True) def _truncate(text: str, limit: int = SLA_TEXT_LIMIT) -> str: if len(text) <= limit: return text return text[:limit] + "\n...[truncated]" def _normalize_addr(raw: str) -> str: s = str(raw or "").strip().lower() if s.startswith("0x") and len(s) == 42: return s return s def _host_from_url(url: str) -> str: s = str(url or "").strip().lower() s = s.replace("https://", "").replace("http://", "") s = s.split("/")[0].split("?")[0].split("#")[0] if s.startswith("www."): s = s[4:] return s def _host_allowed(host: str, suffixes: tuple) -> bool: h = (host or "").lower().strip() if not h: return False for suffix in suffixes: if h == suffix or h.endswith("." + suffix): return True return False def _as_bool(value) -> bool: if value is True: return True if value is False: return False s = str(value or "").strip().lower() return s in ("true", "1", "yes") def _fingerprint(final_host: str, sla_title: str, sla_text: str) -> str: blob = "|".join( [ (final_host or "").lower().strip(), (sla_title or "").lower().strip()[:160], (sla_text or "").lower().strip()[:4000], ] ) acc = 0 for i, ch in enumerate(blob): acc = (acc * 131 + ord(ch) + i) % 1_000_000_000_007 return f"{acc}:{blob[:180]}" @gl.evm.contract_interface class _Recipient: class View: pass class Write: pass @allow_storage @dataclass class Credit: id: u256 provider: Address customer: Address sla_url: str incident_url: str memo: str amount: u256 status: str created_at: u256 funded_at: u256 settled_at: u256 deadline: u256 claim_window_seconds: u256 ai_summary: str sla_title: str covered_services: str sla_text: str sla_fingerprint: str sla_final_url: str sla_final_host: str last_eval_summary: str settle_reason: str paid_out: u256 class IncidentCredit(gl.Contract): credits: TreeMap[u256, Credit] credit_count: u256 def __init__(self): self.credit_count = u256(0) def _now_epoch(self) -> u256: try: from datetime import datetime, timezone return u256(int(datetime.now(timezone.utc).timestamp())) except Exception: pass try: import time as _time return u256(int(_time.time())) except Exception: pass try: raw = gl.message_raw.get("datetime") if raw: from datetime import datetime text = str(raw).replace("Z", "+00:00") return u256(int(datetime.fromisoformat(text).timestamp())) except Exception: pass return u256(1_788_000_000 + int(self.credit_count)) def _to_dict(self, c: Credit) -> dict: return { "id": int(c.id), "provider": c.provider.as_hex, "customer": c.customer.as_hex, "sla_url": c.sla_url, "incident_url": c.incident_url, "memo": c.memo, "amount": int(c.amount), "status": c.status, "created_at": int(c.created_at), "funded_at": int(c.funded_at), "settled_at": int(c.settled_at), "deadline": int(c.deadline), "claim_window_seconds": int(c.claim_window_seconds), "ai_summary": c.ai_summary, "sla_title": c.sla_title, "covered_services": c.covered_services, "sla_text": c.sla_text, "sla_fingerprint": c.sla_fingerprint, "sla_final_url": c.sla_final_url, "sla_final_host": c.sla_final_host, "last_eval_summary": c.last_eval_summary, "settle_reason": c.settle_reason, "paid_out": int(c.paid_out) == 1, } def _crawl_url_strict(self, url: str) -> str: def fetch_page(): return gl.nondet.web.render(url, mode="text") return gl.eq_principle.strict_eq(fetch_page) def _parse_verdict_json(self, raw: str) -> dict: try: return json.loads(raw) except json.JSONDecodeError: return {} def _assert_sla_host(self, url: str) -> str: host = _host_from_url(url) if not _host_allowed(host, SLA_HOST_SUFFIXES): raise gl.vm.UserError( "SLA host not allowlisted. Use cloudflare.com or example.com." ) return host def _assert_incident_host(self, url: str) -> str: host = _host_from_url(url) if not _host_allowed(host, INCIDENT_HOST_SUFFIXES): raise gl.vm.UserError( "Incident host not allowlisted. Use cloudflarestatus.com or example.com." ) return host def _normalize_window(self, claim_window_seconds: u256) -> int: window = int(claim_window_seconds) if window < 0: raise gl.vm.UserError("claim_window_seconds must be >= 0") if window == 0: return DEFAULT_CLAIM_WINDOW if window > MAX_CLAIM_WINDOW: raise gl.vm.UserError("claim_window_seconds exceeds 90 days") return window def _validate_sla_verdict_shape(self, verdict: dict) -> bool: valid = verdict.get("valid") if valid is not True and valid is not False: return False if not valid: return len(str(verdict.get("reject_reason") or "").strip()) >= 10 if not _as_bool(verdict.get("is_sla_document")): return False if len(str(verdict.get("evidence_summary") or "").strip()) < 15: return False if len(str(verdict.get("sla_title") or "").strip()) < 3: return False if len(str(verdict.get("covered_services") or "").strip()) < 3: return False host = str(verdict.get("final_host") or "").strip().lower() if not _host_allowed(host, SLA_HOST_SUFFIXES): return False return True def _sla_verdict_agrees(self, leader_d: dict, validator_d: dict) -> bool: if leader_d.get("valid") != validator_d.get("valid"): return False if not leader_d.get("valid"): return bool(str(leader_d.get("reject_reason") or "").strip()) == bool( str(validator_d.get("reject_reason") or "").strip() ) return ( _as_bool(leader_d.get("is_sla_document")) == _as_bool(validator_d.get("is_sla_document")) and str(leader_d.get("final_host") or "").strip().lower() == str(validator_d.get("final_host") or "").strip().lower() and bool(str(leader_d.get("sla_title") or "").strip()) == bool(str(validator_d.get("sla_title") or "").strip()) ) def _validate_incident_verdict_shape(self, verdict: dict) -> bool: required = ( "accepted", "is_incident_not_maintenance", "covered_service_affected", "meets_outage_definition", "is_incident_report", ) for key in required: if verdict.get(key) is not True and verdict.get(key) is not False: return False if not verdict.get("accepted"): return len(str(verdict.get("reject_reason") or "").strip()) >= 8 if len(str(verdict.get("evidence_summary") or "").strip()) < 15: return False if not _as_bool(verdict.get("is_incident_report")): return False host = str(verdict.get("final_host") or "").strip().lower() if not _host_allowed(host, INCIDENT_HOST_SUFFIXES): return False return True def _incident_verdict_agrees(self, leader_d: dict, validator_d: dict) -> bool: keys = ( "accepted", "is_incident_not_maintenance", "covered_service_affected", "meets_outage_definition", "is_incident_report", ) for key in keys: if _as_bool(leader_d.get(key)) != _as_bool(validator_d.get(key)): return False return ( str(leader_d.get("final_host") or "").strip().lower() == str(validator_d.get("final_host") or "").strip().lower() ) def _ai_pin_sla(self, memo: str, sla_url: str, crawled_blob: str) -> dict: input_host = _host_from_url(sla_url) def leader_fn() -> str: task = f""" IncidentCredit SLA pin. You verify a public service-level agreement (SLA) document for IncidentCredit. Goals: 1) Decide if this page is a real SLA / service-credit / uptime legal policy, not a status incident, blog, or marketing page. 2) Extract the SLA title and which services it covers. 3) Report the final host after any redirects. 4) Reject 404, empty, unrelated, or non-SLA pages. 5) Do not extract any wallet. Do not decide payouts. Provider memo: {memo[:2000]} SLA URL: {sla_url} URL host: {input_host} Crawled page: {crawled_blob[:11000]} Respond ONLY with JSON: {{ "valid": true, "is_sla_document": true, "evidence_summary": "factual description of the SLA and covered services", "reject_reason": null, "sla_title": "title as printed", "covered_services": "services / products named in the SLA", "final_host": "hostname after redirects, e.g. cloudflare.com", "final_url": "final canonical sla url if known", "policy_excerpt": "short excerpt of the credit / outage definition" }} """ result = gl.nondet.exec_prompt(task, response_format="json") data = _parse_llm_json(result) if data.get("valid") is True: if not str(data.get("final_host") or "").strip(): data["final_host"] = input_host if not str(data.get("final_url") or "").strip(): data["final_url"] = sla_url if "is_sla_document" not in data: data["is_sla_document"] = True return _stable_json(data) def validator_fn(leader_result) -> bool: if not isinstance(leader_result, glvm.Return): return False if not isinstance(leader_result.calldata, str): return False leader_d = self._parse_verdict_json(leader_result.calldata) if not self._validate_sla_verdict_shape(leader_d): return False own_d = self._parse_verdict_json(leader_fn()) if not self._validate_sla_verdict_shape(own_d): return False return self._sla_verdict_agrees(leader_d, own_d) raw = glvm.run_nondet_unsafe(leader_fn, validator_fn) verdict = json.loads(raw) if not self._validate_sla_verdict_shape(verdict): raise gl.vm.UserError("AI SLA verification failed structural validation") return verdict def _ai_evaluate_incident( self, memo: str, incident_url: str, crawled_blob: str, pinned_sla_text: str, sla_title: str, covered_services: str, ) -> dict: input_host = _host_from_url(incident_url) def leader_fn() -> str: task = f""" IncidentCredit incident claim. You compare a public incident narrative against a pinned SLA for IncidentCredit. You are a parser, not a financial judge. Read the incident page and the pinned SLA text. Do not call APIs. Do not use JSON status endpoints. Use only the crawled page text. Decide these facts: - is_incident_report: this page is a public incident / outage narrative, not marketing - is_incident_not_maintenance: this is an unplanned incident, not scheduled maintenance - covered_service_affected: a service named in the pinned SLA is actually affected - meets_outage_definition: the narrative meets the pinned SLA's outage / credit definition Set accepted true only if ALL four of those are true. Provider memo: {memo[:2000]} Incident URL: {incident_url} URL host: {input_host} Pinned SLA title: {sla_title[:240]} Pinned covered services: {covered_services[:500]} Pinned SLA text: {pinned_sla_text[:8000]} Crawled incident page: {crawled_blob[:11000]} Respond ONLY with JSON: {{ "accepted": true, "is_incident_report": true, "is_incident_not_maintenance": true, "covered_service_affected": true, "meets_outage_definition": true, "evidence_summary": "factual comparison of incident vs pinned SLA", "reject_reason": null, "incident_title": "incident title as printed", "final_host": "hostname after redirects, e.g. cloudflarestatus.com", "final_url": "final canonical incident url if known" }} """ result = gl.nondet.exec_prompt(task, response_format="json") data = _parse_llm_json(result) if not str(data.get("final_host") or "").strip(): data["final_host"] = input_host if not str(data.get("final_url") or "").strip(): data["final_url"] = incident_url return _stable_json(data) def validator_fn(leader_result) -> bool: if not isinstance(leader_result, glvm.Return): return False if not isinstance(leader_result.calldata, str): return False leader_d = self._parse_verdict_json(leader_result.calldata) if not self._validate_incident_verdict_shape(leader_d): return False own_d = self._parse_verdict_json(leader_fn()) if not self._validate_incident_verdict_shape(own_d): return False return self._incident_verdict_agrees(leader_d, own_d) raw = glvm.run_nondet_unsafe(leader_fn, validator_fn) verdict = json.loads(raw) if not self._validate_incident_verdict_shape(verdict): raise gl.vm.UserError("AI incident evaluation failed structural validation") return verdict def _qualifies(self, verdict: dict) -> bool: return ( _as_bool(verdict.get("accepted")) and _as_bool(verdict.get("is_incident_report")) and _as_bool(verdict.get("is_incident_not_maintenance")) and _as_bool(verdict.get("covered_service_affected")) and _as_bool(verdict.get("meets_outage_definition")) ) def _payout(self, c: Credit, to_customer: bool, reason: str) -> None: if int(c.paid_out) == 1: raise gl.vm.UserError("Credit already settled") amount = c.amount recipient = c.customer if to_customer else c.provider c.paid_out = u256(1) c.amount = u256(0) c.status = "RELEASED" if to_customer else "REFUNDED" c.settle_reason = reason[:500] c.settled_at = self._now_epoch() if amount > 0: _Recipient(recipient).emit_transfer(value=amount) @gl.public.write def init_credit( self, sla_url: str, customer: str, memo: str, claim_window_seconds: u256, ) -> int: """Step 1: pin SLA text and bind customer. No GEN locked.""" url = str(sla_url or "").strip() memo_text = str(memo or "").strip() cust = _normalize_addr(customer) if not url: raise gl.vm.UserError("sla_url is required") if not memo_text or len(memo_text) < 10: raise gl.vm.UserError("memo must be at least 10 characters") if not (cust.startswith("0x") and len(cust) == 42): raise gl.vm.UserError("customer must be a 0x-prefixed 20-byte address") if gl.message.sender_address == Address(cust): raise gl.vm.UserError("You cannot open an incident credit to yourself") window = self._normalize_window(u256(int(claim_window_seconds))) self._assert_sla_host(url) crawled = self._crawl_url_strict(url) sla_text = _truncate(crawled) verdict = self._ai_pin_sla(memo_text, url, sla_text) if not verdict.get("valid"): raise gl.vm.UserError( verdict.get("reject_reason") or "SLA URL failed AI verification" ) if not _as_bool(verdict.get("is_sla_document")): raise gl.vm.UserError("Page is not an SLA document") final_host = str(verdict.get("final_host") or _host_from_url(url)).strip().lower() if not _host_allowed(final_host, SLA_HOST_SUFFIXES): raise gl.vm.UserError("Final SLA host after redirects is not allowlisted") sla_title = str(verdict.get("sla_title") or "").strip()[:240] covered = str(verdict.get("covered_services") or "").strip()[:500] final_url = str(verdict.get("final_url") or url).strip() fp = _fingerprint(final_host, sla_title, sla_text) cid = self.credit_count self.credit_count = u256(int(self.credit_count) + 1) now = self._now_epoch() self.credits[cid] = Credit( id=cid, provider=gl.message.sender_address, customer=Address(cust), sla_url=url[:500], incident_url="", memo=memo_text[:2000], amount=u256(0), status="PENDING_FUND", created_at=now, funded_at=u256(0), settled_at=u256(0), deadline=u256(0), claim_window_seconds=u256(window), ai_summary=str(verdict.get("evidence_summary") or "")[:2000], sla_title=sla_title, covered_services=covered, sla_text=sla_text[:SLA_TEXT_LIMIT], sla_fingerprint=fp, sla_final_url=final_url[:500], sla_final_host=final_host[:120], last_eval_summary="", settle_reason="", paid_out=u256(0), ) return int(cid) @gl.public.write.payable def fund_credit(self, credit_id: u256) -> None: """Step 2: lock GEN only after the SLA is pinned.""" if credit_id not in self.credits: raise gl.vm.UserError("Credit not found") c = self.credits[credit_id] if c.status != "PENDING_FUND": raise gl.vm.UserError("Credit is not awaiting funding") if gl.message.sender_address != c.provider: raise gl.vm.UserError("Only the provider can fund") v = gl.message.value if v < u256(MIN_FUND_WEI): raise gl.vm.UserError("Send at least 0.01 GEN to lock the credit") now = self._now_epoch() c.amount = v c.funded_at = now c.deadline = u256(int(now) + int(c.claim_window_seconds)) c.status = "FUNDED" @gl.public.write def cancel_credit(self, credit_id: u256) -> None: if credit_id not in self.credits: raise gl.vm.UserError("Credit not found") c = self.credits[credit_id] if c.status != "PENDING_FUND": raise gl.vm.UserError("Only unfunded credits can be cancelled") if gl.message.sender_address != c.provider: raise gl.vm.UserError("Only the provider can cancel") c.status = "CANCELLED" @gl.public.write def evaluate_claim(self, credit_id: u256, incident_url: str) -> None: """Pay the bound customer only if a later incident qualifies under the pinned SLA.""" if credit_id not in self.credits: raise gl.vm.UserError("Credit not found") c = self.credits[credit_id] if c.status != "FUNDED": raise gl.vm.UserError("Credit is not funded") now = int(self._now_epoch()) if now >= int(c.deadline): raise gl.vm.UserError("Claim window expired. Use expire_credit.") url = str(incident_url or "").strip() if not url: raise gl.vm.UserError("incident_url is required") self._assert_incident_host(url) c.incident_url = url[:500] try: crawled = self._crawl_url_strict(url) except Exception: c.last_eval_summary = "Incident page could not be fetched; credit stays funded for retry" raise gl.vm.UserError( "Incident URL failed to load. Credit stays funded. Retry with a live public incident page." ) verdict = self._ai_evaluate_incident( c.memo, url, _truncate(crawled), c.sla_text, c.sla_title, c.covered_services, ) c.last_eval_summary = str( verdict.get("evidence_summary") or verdict.get("reject_reason") or "" )[:2000] final_host = str(verdict.get("final_host") or _host_from_url(url)).strip().lower() if not _host_allowed(final_host, INCIDENT_HOST_SUFFIXES): raise gl.vm.UserError( "Final incident host after redirects is not allowlisted. Credit stays funded." ) if not self._qualifies(verdict): reason = str(verdict.get("reject_reason") or "").strip() if not reason: reason = "Incident does not qualify under the pinned SLA" raise gl.vm.UserError( reason + " Credit stays funded so another incident URL can be evaluated." ) self._payout( c, True, str(verdict.get("evidence_summary") or "Incident qualifies under pinned SLA"), ) @gl.public.write def expire_credit(self, credit_id: u256) -> None: """Refund the provider after the claim window with no qualifying incident.""" if credit_id not in self.credits: raise gl.vm.UserError("Credit not found") c = self.credits[credit_id] if c.status != "FUNDED": raise gl.vm.UserError("Credit is not funded") if int(self._now_epoch()) < int(c.deadline): raise gl.vm.UserError("Claim window is still open") self._payout(c, False, "Claim window expired; unused credit returned to provider") @gl.public.view def get_credit(self, credit_id: u256) -> dict: if credit_id not in self.credits: raise gl.vm.UserError("Credit not found") return self._to_dict(self.credits[credit_id]) @gl.public.view def get_credits_page(self, offset: u256, limit: u256) -> list: off = int(offset) lim = int(limit) if off < 0: raise gl.vm.UserError("offset must be >= 0") if lim < 1 or lim > PAGE_LIMIT: raise gl.vm.UserError("limit must be 1-50") total = int(self.credit_count) out = [] i = off while i < total and len(out) < lim: cid = u256(i) if cid in self.credits: out.append(self._to_dict(self.credits[cid])) i += 1 return out @gl.public.view def get_config(self) -> dict: return { "credit_count": int(self.credit_count), "sla_host_suffixes": list(SLA_HOST_SUFFIXES), "incident_host_suffixes": list(INCIDENT_HOST_SUFFIXES), "page_limit": PAGE_LIMIT, "min_fund_wei": MIN_FUND_WEI, "default_claim_window_seconds": DEFAULT_CLAIM_WINDOW, }