# { "Depends": "py-genlayer:1jb45aa8ynh2a9c9xn3b7qqh8sm5q93hwfp7jqmwsfhh8jpz09h6" } import hashlib import json from datetime import datetime, timezone from genlayer import * STATUS_ACTIVE = "active" STATUS_PAUSED = "paused" STATUS_BREACHED = "breached" STATUS_CLEAR = "clear" STATUS_UNAVAILABLE = "unavailable" ERROR_EXPECTED = "[EXPECTED]" ERROR_EXTERNAL = "[EXTERNAL]" ERROR_TRANSIENT = "[TRANSIENT]" # Tolerated drift between configured window and actual period length (seconds) WINDOW_TOLERANCE_SECONDS = 600 # Window duration bounds MIN_WINDOW_SECONDS = 600 # 10 minutes minimum MAX_WINDOW_SECONDS = 31536000 # 365 days maximum # Check cadence constants: six measurement intervals per 24 hours. A complete # window includes both boundary observations, so a 24-hour window has 7 checks. BASE_CHECK_INTERVAL_SECONDS = 14400 # 4 hours per check -> 6 checks in 24h MIN_CHECK_INTERVAL_SECONDS = 60 # Physical minimum interval between checks (1 minute) MIN_ATTESTATIONS = 2 MAX_ATTESTATIONS = 9 class Pactline(gl.Contract): """Provider service registry, SLA coverage pool, evidence ledger, and settlement engine.""" owner: Address monitor_operator: Address authorized_monitors: TreeMap[str, bool] service_seq: u256 subscription_seq: u256 snapshot_seq: u256 attestation_seq: u256 claim_seq: u256 credit_seq: u256 services: TreeMap[str, str] service_order: DynArray[str] subscriptions: TreeMap[str, str] subscription_order: DynArray[str] subscription_index: TreeMap[str, str] snapshots: TreeMap[str, str] snapshot_order: DynArray[str] attestations: TreeMap[str, str] attestation_order: DynArray[str] claims: TreeMap[str, str] claim_order: DynArray[str] claimed_windows: TreeMap[str, bool] credits: TreeMap[str, str] credit_order: DynArray[str] def __init__(self): self.owner = gl.message.sender_address self.monitor_operator = gl.message.sender_address self.authorized_monitors[self._addr_key(gl.message.sender_address)] = True self.service_seq = u256(0) self.subscription_seq = u256(0) self.snapshot_seq = u256(0) self.attestation_seq = u256(0) self.claim_seq = u256(0) self.credit_seq = u256(0) def _addr_key(self, value) -> str: if isinstance(value, bytes): return "0x" + value.hex().lower() text = str(value).strip().lower() if hasattr(value, "as_hex"): return str(value.as_hex).lower() return text def _now_epoch(self) -> u256: try: raw = getattr(gl, "message_raw", {}) if isinstance(raw, dict) and raw.get("datetime"): value = str(raw["datetime"]).replace("Z", "+00:00") parsed = datetime.fromisoformat(value) if parsed.tzinfo is None: parsed = parsed.replace(tzinfo=timezone.utc) return u256(int(parsed.timestamp())) return u256(int(datetime.now(timezone.utc).timestamp())) except Exception: return u256(0) def _date_epoch(self, value: str) -> u256: try: parsed = datetime.fromisoformat(str(value).replace("Z", "+00:00")) if parsed.tzinfo is None: parsed = parsed.replace(tzinfo=timezone.utc) return u256(int(parsed.timestamp())) except Exception: raise gl.vm.UserError(ERROR_EXPECTED + " period end is invalid") def _service_key(self, service_id: u256) -> str: return str(int(service_id)) def _snapshot_key(self, service_id: str, period_start: str) -> str: digest = hashlib.sha256( f"{service_id}:{period_start}".encode("utf-8") ).hexdigest() return "snapshot_" + digest[:48] def _subscription_key(self, service_id: str, customer: str) -> str: digest = hashlib.sha256( f"{service_id}:{customer}".encode("utf-8") ).hexdigest() return "subscription_index_" + digest[:48] def _claim_key(self, subscription_id: str, snapshot_id: str) -> str: digest = hashlib.sha256( f"{subscription_id}:{snapshot_id}".encode("utf-8") ).hexdigest() return "claim_window_" + digest[:48] def _period_claim_key(self, subscription_id: str, period_start: str) -> str: digest = hashlib.sha256( f"{subscription_id}:{period_start}".encode("utf-8") ).hexdigest() return "claim_period_" + digest[:48] def _load_service(self, service_id: str) -> dict: key = str(service_id).strip() encoded = self.services.get(key, "") if not encoded: raise gl.vm.UserError(ERROR_EXPECTED + " service does not exist") return json.loads(encoded) def _load_subscription(self, subscription_id: str) -> dict: key = str(subscription_id).strip() encoded = self.subscriptions.get(key, "") if not encoded: raise gl.vm.UserError(ERROR_EXPECTED + " subscription does not exist") return json.loads(encoded) def _load_snapshot(self, snapshot_id: str) -> dict: encoded = self.snapshots.get(str(snapshot_id).strip(), "") if not encoded: raise gl.vm.UserError(ERROR_EXPECTED + " snapshot does not exist") return json.loads(encoded) def _load_attestation(self, attestation_id: str) -> dict: encoded = self.attestations.get(str(attestation_id).strip(), "") if not encoded: raise gl.vm.UserError(ERROR_EXPECTED + " attestation does not exist") return json.loads(encoded) def _validate_url(self, value: str, field: str) -> str: url = str(value).strip() if not (url.startswith("https://") or url.startswith("http://")): raise gl.vm.UserError(ERROR_EXPECTED + f" {field} must be an http URL") if len(url) > 512: raise gl.vm.UserError(ERROR_EXPECTED + f" {field} is too long") return url def _validate_date(self, value: str, field: str) -> str: text = str(value).strip() try: datetime.fromisoformat(text.replace("Z", "+00:00")) except Exception: raise gl.vm.UserError( ERROR_EXPECTED + f" {field} must be an ISO date" ) return text def _max_payout(self, service: dict, subscription_fee: int) -> u256: return ( u256(subscription_fee) * u256(int(service["compensation_bps"])) // u256(10000) ) def _resolve_evidence(self, evidence_url: str) -> dict: """Fetch one monitor packet and normalize only stable measurement fields.""" url = self._validate_url(evidence_url, "evidence_url") def fetch_and_normalize() -> dict: try: response = gl.nondet.web.get( url, headers={"Accept": "application/json"}, ) if response.status >= 500: raise gl.vm.UserError( ERROR_TRANSIENT + f" evidence returned {response.status}" ) if response.status >= 400: raise gl.vm.UserError( ERROR_EXTERNAL + f" evidence returned {response.status}" ) body = response.body if isinstance(body, str): body = body.encode("utf-8") parsed = json.loads(body.decode("utf-8", errors="replace")) if not isinstance(parsed, dict): raise gl.vm.UserError( ERROR_EXTERNAL + " evidence is not a JSON object" ) checks = parsed.get("checks", []) if not isinstance(checks, list): raise gl.vm.UserError( ERROR_EXTERNAL + " evidence checks are not an array" ) normalized_checks = [] for item in checks: if not isinstance(item, dict): raise gl.vm.UserError( ERROR_EXTERNAL + " evidence contains an invalid check" ) normalized_checks.append( { "slot": int(item.get("slot", -1)), "checked_at": str(item.get("checked_at", "")), "ok": bool(item.get("ok", False)), "response_code": int(item.get("response_code", 0)), } ) return { "service_id": str(parsed.get("service_id", "")), "period_start": str(parsed.get("period_start", "")), "period_end": str(parsed.get("period_end", "")), "monitor_address": str( parsed.get("monitor_address", "") ).strip().lower(), "checks": normalized_checks, "signature": str(parsed.get("signature", "")), } except gl.vm.UserError: raise except Exception: raise gl.vm.UserError( ERROR_EXTERNAL + " evidence could not be parsed" ) return gl.eq_principle.strict_eq(fetch_and_normalize) def _canonical_evidence_without_signature(self, evidence: dict) -> str: return json.dumps( { "service_id": evidence["service_id"], "period_start": evidence["period_start"], "period_end": evidence["period_end"], "monitor_address": evidence["monitor_address"], "checks": evidence["checks"], }, sort_keys=True, separators=(",", ":"), ) def _validate_packet( self, evidence: dict, service: dict, period_start: str, period_end: str, expected_seconds: int, ) -> None: if evidence["service_id"] != str(service["service_id"]): raise gl.vm.UserError( ERROR_EXTERNAL + " evidence service_id does not match" ) if evidence["period_start"] != period_start: raise gl.vm.UserError( ERROR_EXTERNAL + " evidence period_start does not match" ) if evidence["period_end"] != period_end: raise gl.vm.UserError( ERROR_EXTERNAL + " evidence period_end does not match" ) checks = evidence["checks"] expected_checks, _, max_checks = self._derive_check_bounds(expected_seconds) if len(checks) != expected_checks: raise gl.vm.UserError( ERROR_EXTERNAL + f" evidence must contain exactly {expected_checks} checks" ) if not evidence["monitor_address"].startswith("0x"): raise gl.vm.UserError(ERROR_EXTERNAL + " monitor address is invalid") digest = hashlib.sha256( self._canonical_evidence_without_signature(evidence).encode("utf-8") ).hexdigest() if evidence["signature"] != digest: raise gl.vm.UserError( ERROR_EXTERNAL + " evidence content hash does not match" ) start_epoch = int(self._date_epoch(period_start)) end_epoch = int(self._date_epoch(period_end)) published_epoch = int(self._now_epoch()) previous_epoch = None for index, check in enumerate(checks): if int(check["slot"]) != index: raise gl.vm.UserError( ERROR_EXTERNAL + " evidence check slots are invalid" ) checked_at = self._validate_date(check["checked_at"], "checked_at") checked_epoch = int(self._date_epoch(checked_at)) if checked_epoch < start_epoch or checked_epoch > end_epoch: raise gl.vm.UserError( ERROR_EXTERNAL + " check timestamp is outside the SLA window" ) if checked_epoch > published_epoch + WINDOW_TOLERANCE_SECONDS: raise gl.vm.UserError( ERROR_EXTERNAL + " check timestamp is in the future" ) if previous_epoch is not None: if checked_epoch <= previous_epoch: raise gl.vm.UserError( ERROR_EXTERNAL + " check timestamps must be increasing" ) if checked_epoch - previous_epoch < MIN_CHECK_INTERVAL_SECONDS: raise gl.vm.UserError( ERROR_EXTERNAL + " checks are too close together" ) previous_epoch = checked_epoch first_epoch = int(self._date_epoch(checks[0]["checked_at"])) last_epoch = int(self._date_epoch(checks[-1]["checked_at"])) if first_epoch - start_epoch > WINDOW_TOLERANCE_SECONDS: raise gl.vm.UserError( ERROR_EXTERNAL + " first check does not cover the window start" ) if end_epoch - last_epoch > WINDOW_TOLERANCE_SECONDS: raise gl.vm.UserError( ERROR_EXTERNAL + " last check does not cover the window end" ) minimum_span = expected_seconds - (2 * WINDOW_TOLERANCE_SECONDS) if last_epoch - first_epoch < minimum_span: raise gl.vm.UserError( ERROR_EXTERNAL + " checks do not span the configured SLA window" ) for index in range(1, len(checks)): left = int(self._date_epoch(checks[index - 1]["checked_at"])) right = int(self._date_epoch(checks[index]["checked_at"])) if right - left < ( expected_seconds // (expected_checks - 1) ) - WINDOW_TOLERANCE_SECONDS: raise gl.vm.UserError( ERROR_EXTERNAL + " check cadence is compressed" ) if len(checks) > max_checks: raise gl.vm.UserError(ERROR_EXTERNAL + " evidence has too many checks") for check in checks: if int(check["response_code"]) < 0 or int(check["response_code"]) > 599: raise gl.vm.UserError( ERROR_EXTERNAL + " response code is invalid" ) def _collect_attestations( self, service: dict, period_start: str, period_end: str, attestation_ids: list, recheck: bool = True, ) -> tuple[dict, list]: if len(attestation_ids) < MIN_ATTESTATIONS: raise gl.vm.UserError( ERROR_EXPECTED + f" at least {MIN_ATTESTATIONS} monitor attestations are required" ) if len(attestation_ids) > MAX_ATTESTATIONS: raise gl.vm.UserError(ERROR_EXPECTED + " too many attestations") expected_seconds = int( service.get("window_seconds", service["window_days"] * 86400) ) packets = [] seen_monitors = {} for raw_id in attestation_ids: attestation = self._load_attestation(str(raw_id)) if attestation["service_id"] != str(service["service_id"]): raise gl.vm.UserError( ERROR_EXPECTED + " attestation belongs to another service" ) if attestation["period_start"] != period_start or attestation[ "period_end" ] != period_end: raise gl.vm.UserError( ERROR_EXPECTED + " attestation period does not match" ) monitor = self._addr_key(attestation["monitor_address"]) if seen_monitors.get(monitor, False): raise gl.vm.UserError( ERROR_EXPECTED + " attestations must use distinct monitors" ) seen_monitors[monitor] = True packet = ( self._resolve_evidence(attestation["evidence_url"]) if recheck else json.loads(attestation["evidence"]) ) if packet != json.loads(attestation["evidence"]): raise gl.vm.UserError( ERROR_EXTERNAL + " monitor evidence changed after attestation" ) if packet["monitor_address"] != monitor: raise gl.vm.UserError( ERROR_EXTERNAL + " evidence monitor does not match attestation" ) self._validate_packet( packet, service, period_start, period_end, expected_seconds ) packets.append(packet) baseline = packets[0] for packet in packets[1:]: if len(packet["checks"]) != len(baseline["checks"]): raise gl.vm.UserError( ERROR_EXTERNAL + " monitor check counts do not agree" ) for left, right in zip(baseline["checks"], packet["checks"]): left_epoch = int(self._date_epoch(left["checked_at"])) right_epoch = int(self._date_epoch(right["checked_at"])) if ( left["slot"] != right["slot"] or left["ok"] != right["ok"] or abs(left_epoch - right_epoch) > WINDOW_TOLERANCE_SECONDS ): raise gl.vm.UserError( ERROR_EXTERNAL + " independent monitor checks do not agree" ) return baseline, packets def _send_value(self, recipient: Address, amount: u256) -> None: if amount > u256(0): gl.get_contract_at(recipient).emit_transfer( value=amount, on="finalized", ) def _derive_window_seconds(self, window_days: u256) -> int: """Derive window duration in seconds from configured window parameter. Values between 1 and 365 are interpreted as days (days * 86400). Values > 365 are interpreted as raw seconds (supporting sub-day / non-24h periods like 3600s for 1 hour or 43200s for 12 hours). """ val = int(window_days) if 1 <= val <= 365: return val * 86400 elif val > 365: if val > MAX_WINDOW_SECONDS: raise gl.vm.UserError( ERROR_EXPECTED + " measurement window exceeds maximum" ) return val else: raise gl.vm.UserError( ERROR_EXPECTED + " measurement window must be positive" ) def _derive_check_bounds(self, window_seconds: int) -> tuple[int, int, int]: """Derive expected checks, minimum checks, and maximum physical checks based on SLA window. - expected_checks: derived from window length / 4-hour base interval (6 checks per 24h). - min_checks: minimum required checks (at least 1 for any window). - max_checks: physical ceiling bounded by MIN_CHECK_INTERVAL_SECONDS (60s). """ expected_checks = max( 2, (window_seconds // BASE_CHECK_INTERVAL_SECONDS) + 1 ) min_checks = expected_checks max_checks = max(1, window_seconds // MIN_CHECK_INTERVAL_SECONDS) return expected_checks, min_checks, max_checks @gl.public.view def get_sla_window_params(self, service_id: str) -> str: """Return derived window duration and check count bounds for a service.""" service = self._load_service(service_id) w_sec = int(service.get("window_seconds", service["window_days"] * 86400)) exp, min_c, max_c = self._derive_check_bounds(w_sec) return json.dumps( { "window_seconds": w_sec, "expected_checks": exp, "min_checks": min_c, "max_checks": max_c, "base_interval_seconds": BASE_CHECK_INTERVAL_SECONDS, "min_interval_seconds": MIN_CHECK_INTERVAL_SECONDS, "minimum_attestations": MIN_ATTESTATIONS, }, sort_keys=True, ) @gl.public.write def set_monitor_operator(self, operator: Address) -> None: """Set the primary monitor operator and authorize it (backwards-compatible).""" if gl.message.sender_address != self.owner: raise gl.vm.UserError(ERROR_EXPECTED + " only the owner can set the monitor") self.monitor_operator = operator self.authorized_monitors[self._addr_key(operator)] = True @gl.public.write def authorize_monitor(self, monitor: Address) -> None: """Authorize an additional monitor operator for uptime measurement submissions.""" if gl.message.sender_address != self.owner: raise gl.vm.UserError(ERROR_EXPECTED + " only the owner can authorize monitors") self.authorized_monitors[self._addr_key(monitor)] = True @gl.public.write def revoke_monitor(self, monitor: Address) -> None: """Revoke authorization for a monitor operator.""" if gl.message.sender_address != self.owner: raise gl.vm.UserError(ERROR_EXPECTED + " only the owner can revoke monitors") self.authorized_monitors[self._addr_key(monitor)] = False @gl.public.view def is_authorized_monitor(self, monitor: Address) -> bool: """Check if an address is authorized to submit monitor measurements.""" return self.authorized_monitors.get(self._addr_key(monitor), False) @gl.public.write.payable def register_service( self, service_name: str, service_url: str, terms: str, threshold_bps: u256, window_days: u256, compensation_type: str, compensation_bps: u256, subscription_price_wei: u256, ) -> str: name = str(service_name).strip() if not name or len(name) > 96: raise gl.vm.UserError(ERROR_EXPECTED + " service name is required") url = self._validate_url(service_url, "service_url") terms_text = str(terms).strip() if not terms_text or len(terms_text) > 1000: raise gl.vm.UserError(ERROR_EXPECTED + " service terms are required") if threshold_bps < u256(1) or threshold_bps > u256(10000): raise gl.vm.UserError( ERROR_EXPECTED + " threshold must be between 0.01 and 100 percent" ) window_seconds = self._derive_window_seconds(window_days) if window_seconds < MIN_WINDOW_SECONDS or window_seconds > MAX_WINDOW_SECONDS: raise gl.vm.UserError( ERROR_EXPECTED + " measurement window must be between 10 minutes and 365 days" ) kind = str(compensation_type).strip().lower() if kind not in ("refund", "credit"): raise gl.vm.UserError( ERROR_EXPECTED + " compensation must be refund or credit" ) if compensation_bps < u256(1) or compensation_bps > u256(10000): raise gl.vm.UserError( ERROR_EXPECTED + " compensation must be between 0.01 and 100 percent" ) if subscription_price_wei <= u256(0): raise gl.vm.UserError( ERROR_EXPECTED + " subscription price is required" ) if gl.message.value <= u256(0): raise gl.vm.UserError( ERROR_EXPECTED + " provider collateral is required" ) max_payout = self._max_payout( {"compensation_bps": int(compensation_bps)}, int(subscription_price_wei), ) if max_payout <= u256(0) or gl.message.value < max_payout: raise gl.vm.UserError( ERROR_EXPECTED + " collateral must cover one subscription claim" ) self.service_seq += u256(1) service_id = self._service_key(self.service_seq) service = { "service_id": service_id, "provider": str(gl.message.sender_address), "service_name": name, "service_url": url, "terms": terms_text, "threshold_bps": int(threshold_bps), "window_days": int(window_days), "window_seconds": window_seconds, "compensation_type": kind, "compensation_bps": int(compensation_bps), "subscription_price_wei": int(subscription_price_wei), "collateral_wei": int(gl.message.value), "reserved_wei": 0, "provider_revenue_wei": 0, "subscriber_count": 0, "status": STATUS_ACTIVE, "created_at": int(self._now_epoch()), "last_uptime_bps": 10000, } self.services[service_id] = json.dumps(service, sort_keys=True) self.service_order.append(service_id) return service_id @gl.public.write.payable def add_service_collateral(self, service_id: str) -> None: service = self._load_service(service_id) if str(gl.message.sender_address) != service["provider"]: raise gl.vm.UserError( ERROR_EXPECTED + " only the provider can add collateral" ) if gl.message.value <= u256(0): raise gl.vm.UserError(ERROR_EXPECTED + " collateral must be positive") service["collateral_wei"] = int(service["collateral_wei"]) + int( gl.message.value ) self.services[str(service_id)] = json.dumps(service, sort_keys=True) @gl.public.write def pause_service(self, service_id: str) -> None: service = self._load_service(service_id) if str(gl.message.sender_address) != service["provider"]: raise gl.vm.UserError( ERROR_EXPECTED + " only the provider can pause a service" ) if service["status"] not in (STATUS_ACTIVE, STATUS_PAUSED): raise gl.vm.UserError(ERROR_EXPECTED + " service is already closed") service["status"] = ( STATUS_PAUSED if service["status"] == STATUS_ACTIVE else STATUS_ACTIVE ) self.services[str(service_id)] = json.dumps(service, sort_keys=True) @gl.public.write def withdraw_provider_revenue( self, service_id: str, amount_wei: u256 ) -> None: service = self._load_service(service_id) if str(gl.message.sender_address) != service["provider"]: raise gl.vm.UserError( ERROR_EXPECTED + " only the provider can withdraw revenue" ) if amount_wei <= u256(0) or amount_wei > u256( int(service["provider_revenue_wei"]) ): raise gl.vm.UserError(ERROR_EXPECTED + " revenue amount is unavailable") service["provider_revenue_wei"] = int( u256(int(service["provider_revenue_wei"])) - amount_wei ) self.services[str(service_id)] = json.dumps(service, sort_keys=True) self._send_value(gl.message.sender_address, amount_wei) @gl.public.write.payable def subscribe_service(self, service_id: str) -> str: service = self._load_service(service_id) if service["status"] != STATUS_ACTIVE: raise gl.vm.UserError( ERROR_EXPECTED + " service is not accepting subscribers" ) if str(gl.message.sender_address) == service["provider"]: raise gl.vm.UserError( ERROR_EXPECTED + " provider cannot subscribe to its own service" ) price = u256(int(service["subscription_price_wei"])) if gl.message.value != price: raise gl.vm.UserError( ERROR_EXPECTED + " subscription payment does not match the price" ) max_payout = self._max_payout(service, int(price)) available = u256(int(service["collateral_wei"])) - u256( int(service["reserved_wei"]) ) if available < max_payout: raise gl.vm.UserError( ERROR_EXPECTED + " provider collateral is fully reserved" ) index_key = self._subscription_key( str(service["service_id"]), str(gl.message.sender_address) ) existing_id = self.subscription_index.get(index_key, "") if existing_id: existing = self._load_subscription(existing_id) if existing["status"] == STATUS_ACTIVE: raise gl.vm.UserError( ERROR_EXPECTED + " customer already has an active subscription" ) self.subscription_seq += u256(1) subscription_id = str(int(self.subscription_seq)) subscription = { "subscription_id": subscription_id, "service_id": str(service["service_id"]), "customer": str(gl.message.sender_address), "provider": service["provider"], "subscription_price_wei": int(price), "max_payout_wei": int(max_payout), "started_at": int(self._now_epoch()), "status": STATUS_ACTIVE, "claim_count": 0, "compensated_wei": 0, "reserve_backed": True, "window_index": 1, "last_settled_period_end": 0, } self.subscriptions[subscription_id] = json.dumps( subscription, sort_keys=True ) self.subscription_order.append(subscription_id) self.subscription_index[index_key] = subscription_id service["reserved_wei"] = int( u256(int(service["reserved_wei"])) + max_payout ) service["provider_revenue_wei"] = int( u256(int(service["provider_revenue_wei"])) + price ) service["subscriber_count"] = int(service["subscriber_count"]) + 1 self.services[str(service_id)] = json.dumps(service, sort_keys=True) return subscription_id @gl.public.write def publish_attestation( self, service_id: str, period_start: str, period_end: str, evidence_url: str, signature: str, ) -> str: """Record one independently operated monitor's complete window packet.""" service = self._load_service(service_id) monitor = self._addr_key(gl.message.sender_address) if not self.authorized_monitors.get(monitor, False): raise gl.vm.UserError( ERROR_EXPECTED + " only authorized monitors can attest measurements" ) start = self._validate_date(period_start, "period_start") end = self._validate_date(period_end, "period_end") start_epoch = self._date_epoch(start) end_epoch = self._date_epoch(end) if end_epoch <= start_epoch: raise gl.vm.UserError(ERROR_EXPECTED + " period end must be after start") expected_seconds = int( service.get("window_seconds", service["window_days"] * 86400) ) if ( abs(int(end_epoch) - int(start_epoch) - expected_seconds) > WINDOW_TOLERANCE_SECONDS ): raise gl.vm.UserError( ERROR_EXPECTED + " period length does not match the service window" ) url = self._validate_url(evidence_url, "evidence_url") if not str(signature).strip(): raise gl.vm.UserError(ERROR_EXPECTED + " evidence hash is required") attestation_id = ( "attestation_" + hashlib.sha256( f"{service_id}:{start}:{monitor}".encode("utf-8") ).hexdigest()[:48] ) if self.attestations.get(attestation_id, ""): raise gl.vm.UserError(ERROR_EXPECTED + " monitor already attested this window") evidence = self._resolve_evidence(url) if evidence["monitor_address"] != monitor: raise gl.vm.UserError( ERROR_EXTERNAL + " evidence monitor does not match publisher" ) if evidence["signature"] != str(signature).strip(): raise gl.vm.UserError(ERROR_EXTERNAL + " evidence hash does not match") self._validate_packet(evidence, service, start, end, expected_seconds) attestation = { "attestation_id": attestation_id, "service_id": str(service["service_id"]), "period_start": start, "period_end": end, "monitor_address": monitor, "evidence_url": url, "signature": str(signature).strip(), "evidence": json.dumps(evidence, sort_keys=True, separators=(",", ":")), "published_at": int(self._now_epoch()), } self.attestation_seq += u256(1) self.attestations[attestation_id] = json.dumps(attestation, sort_keys=True) self.attestation_order.append(attestation_id) return attestation_id @gl.public.write def publish_snapshot( self, service_id: str, period_start: str, period_end: str, attestation_ids_json: str, ) -> str: """Aggregate multiple independently attested packets into one snapshot.""" service = self._load_service(service_id) start = self._validate_date(period_start, "period_start") end = self._validate_date(period_end, "period_end") try: attestation_ids = json.loads(str(attestation_ids_json)) except Exception: raise gl.vm.UserError(ERROR_EXPECTED + " attestation ids must be JSON") if not isinstance(attestation_ids, list): raise gl.vm.UserError(ERROR_EXPECTED + " attestation ids must be an array") snapshot_id = self._snapshot_key(str(service["service_id"]), start) if self.snapshots.get(snapshot_id, ""): raise gl.vm.UserError(ERROR_EXPECTED + " snapshot already exists") baseline, packets = self._collect_attestations( service, start, end, attestation_ids ) total_checks = len(baseline["checks"]) failed_checks = sum( 1 for check in baseline["checks"] if not bool(check["ok"]) ) uptime_bps = ((total_checks - failed_checks) * 10000) // total_checks snapshot = { "snapshot_id": snapshot_id, "service_id": str(service["service_id"]), "period_start": start, "period_end": end, "uptime_bps": uptime_bps, "total_checks": total_checks, "failed_checks": failed_checks, "attestation_ids": [str(value) for value in attestation_ids], "source_count": len(packets), "source_monitors": [ self._addr_key(self._load_attestation(str(value))["monitor_address"]) for value in attestation_ids ], "published_at": int(self._now_epoch()), } self.snapshot_seq += u256(1) self.snapshots[snapshot_id] = json.dumps(snapshot, sort_keys=True) self.snapshot_order.append(snapshot_id) service["last_uptime_bps"] = uptime_bps self.services[str(service_id)] = json.dumps(service, sort_keys=True) return snapshot_id @gl.public.write def file_claim(self, subscription_id: str, snapshot_id: str) -> str: """File a claim for an SLA breach. Enforces that: 1. Only the subscriber can file a claim. 2. The subscription is active and has an active reserve backing this window. A subscription whose reserve has been released cannot enter or claim in another window until renewed. 3. The snapshot publisher is verified as authorized. 4. The snapshot period aligns with the configured SLA window. 5. Evidence independently confirms the uptime measurements. Upon resolution, the reserve for this window is released unconditionally. """ subscription = self._load_subscription(subscription_id) if str(gl.message.sender_address) != subscription["customer"]: raise gl.vm.UserError( ERROR_EXPECTED + " only the subscriber can file a claim" ) if subscription["status"] != STATUS_ACTIVE: raise gl.vm.UserError(ERROR_EXPECTED + " subscription is not active") service = self._load_service(subscription["service_id"]) snapshot = self._load_snapshot(snapshot_id) if snapshot["service_id"] != str(service["service_id"]): raise gl.vm.UserError( ERROR_EXPECTED + " snapshot belongs to another service" ) if self.claimed_windows.get( self._period_claim_key(subscription_id, snapshot["period_start"]), False ): raise gl.vm.UserError( ERROR_EXPECTED + " this subscription window already has a claim" ) if not subscription.get("reserve_backed", False): raise gl.vm.UserError( ERROR_EXPECTED + " subscription reserve has already been released" ) if self._now_epoch() < self._date_epoch(snapshot["period_end"]): raise gl.vm.UserError(ERROR_EXPECTED + " measurement window has not ended") if self._date_epoch(snapshot["period_end"]) <= u256( int(subscription["started_at"]) ): raise gl.vm.UserError( ERROR_EXPECTED + " snapshot predates the subscription" ) last_settled_end = int(subscription.get("last_settled_period_end", 0)) if last_settled_end and int(self._date_epoch(snapshot["period_start"])) < last_settled_end: raise gl.vm.UserError( ERROR_EXPECTED + " snapshot overlaps an already settled window" ) # Enforce window alignment at claim time too start_epoch = self._date_epoch(snapshot["period_start"]) end_epoch = self._date_epoch(snapshot["period_end"]) actual_seconds = int(end_epoch) - int(start_epoch) expected_seconds = int(service.get("window_seconds", service["window_days"] * 86400)) if abs(actual_seconds - expected_seconds) > WINDOW_TOLERANCE_SECONDS: raise gl.vm.UserError( ERROR_EXPECTED + " snapshot period does not match the service window" ) # Re-fetch every monitor packet. A claim cannot settle from one aggregate # packet or from stale values copied into the snapshot. try: attestation_ids = snapshot["attestation_ids"] except KeyError: raise gl.vm.UserError( ERROR_EXTERNAL + " snapshot has no monitor attestations" ) evidence, packets = self._collect_attestations( service, snapshot["period_start"], snapshot["period_end"], attestation_ids, ) total_checks = len(evidence["checks"]) failed_checks = sum(1 for check in evidence["checks"] if not bool(check["ok"])) uptime_bps = ((total_checks - failed_checks) * 10000) // total_checks if ( uptime_bps != int(snapshot["uptime_bps"]) or total_checks != int(snapshot["total_checks"]) or failed_checks != int(snapshot["failed_checks"]) ): raise gl.vm.UserError( ERROR_EXTERNAL + " fetched attestations do not match the snapshot" ) breached = uptime_bps < int(service["threshold_bps"]) max_payout = u256(int(subscription["max_payout_wei"])) payout = max_payout if breached else u256(0) if breached and u256(int(service["collateral_wei"])) < payout: raise gl.vm.UserError( ERROR_EXPECTED + " provider collateral is insufficient for this claim" ) # Release the reservation unconditionally — each window is settled exactly once service["reserved_wei"] = int( u256(int(service["reserved_wei"])) - max_payout ) # On breach: deduct payout from collateral regardless of compensation type if payout > u256(0): service["collateral_wei"] = int( u256(int(service["collateral_wei"])) - payout ) self.services[str(service["service_id"])] = json.dumps( service, sort_keys=True ) self.claim_seq += u256(1) claim_id = str(int(self.claim_seq)) status = STATUS_BREACHED if breached else STATUS_CLEAR claim = { "claim_id": claim_id, "service_id": str(service["service_id"]), "subscription_id": str(subscription["subscription_id"]), "customer": subscription["customer"], "provider": service["provider"], "snapshot_id": str(snapshot_id), "status": status, "uptime_bps": uptime_bps, "threshold_bps": int(service["threshold_bps"]), "settlement_type": service["compensation_type"], "settlement_wei": int(payout), "attestation_ids": attestation_ids, "source_count": len(packets), "resolved_at": int(self._now_epoch()), } self.claims[claim_id] = json.dumps(claim, sort_keys=True) self.claim_order.append(claim_id) self.claimed_windows[ self._period_claim_key(subscription_id, snapshot["period_start"]) ] = True subscription["reserve_backed"] = False subscription["last_settled_period_end"] = int( self._date_epoch(snapshot["period_end"]) ) subscription["claim_count"] = int(subscription["claim_count"]) + 1 subscription["compensated_wei"] = int( u256(int(subscription["compensated_wei"])) + payout ) self.subscriptions[str(subscription_id)] = json.dumps( subscription, sort_keys=True ) if payout > u256(0): if service["compensation_type"] == "refund": # Immediate refund transfer self._send_value(gl.message.sender_address, payout) else: # Credit type: create a redeemable credit record. # Collateral has already been deducted above; the customer # calls redeem_credit() to receive the funds. self.credit_seq += u256(1) credit_id = str(int(self.credit_seq)) credit = { "credit_id": credit_id, "claim_id": claim_id, "service_id": str(service["service_id"]), "subscription_id": str(subscription["subscription_id"]), "customer": subscription["customer"], "provider": service["provider"], "settlement_wei": int(payout), "redeemed": False, "created_at": int(self._now_epoch()), "redeemed_at": None, } self.credits[credit_id] = json.dumps(credit, sort_keys=True) self.credit_order.append(credit_id) return claim_id @gl.public.write def redeem_credit(self, credit_id: str) -> None: """Redeem a credit created from a breached SLA claim. Only the customer who earned the credit can redeem it. Each credit may only be redeemed once — double redemption is rejected. The collateral has already been debited when the claim was filed; calling this function transfers the held amount to the customer. """ encoded = self.credits.get(str(credit_id).strip(), "") if not encoded: raise gl.vm.UserError(ERROR_EXPECTED + " credit does not exist") credit = json.loads(encoded) if str(gl.message.sender_address) != credit["customer"]: raise gl.vm.UserError( ERROR_EXPECTED + " only the credit owner can redeem it" ) if credit["redeemed"]: raise gl.vm.UserError( ERROR_EXPECTED + " credit has already been redeemed" ) credit["redeemed"] = True credit["redeemed_at"] = int(self._now_epoch()) self.credits[str(credit_id)] = json.dumps(credit, sort_keys=True) self._send_value(gl.message.sender_address, u256(int(credit["settlement_wei"]))) @gl.public.write.payable def renew_subscription(self, subscription_id: str) -> None: """Renew a subscription for the next window by re-funding provider collateral reserve. A subscription whose reserve was released after claim settlement cannot enter a new window until renewed. The subscriber sends the subscription fee, which re-reserves max_payout from provider collateral and marks the subscription as reserve-backed for the new window. """ subscription = self._load_subscription(subscription_id) if str(gl.message.sender_address) != subscription["customer"]: raise gl.vm.UserError( ERROR_EXPECTED + " only the subscriber can renew the subscription" ) if subscription.get("reserve_backed", False): raise gl.vm.UserError( ERROR_EXPECTED + " subscription already has an active reserve for the current window" ) service = self._load_service(subscription["service_id"]) if service["status"] != STATUS_ACTIVE: raise gl.vm.UserError( ERROR_EXPECTED + " service is not accepting subscribers" ) price = u256(int(service["subscription_price_wei"])) if gl.message.value != price: raise gl.vm.UserError( ERROR_EXPECTED + " subscription payment does not match the price" ) max_payout = self._max_payout(service, int(price)) available = u256(int(service["collateral_wei"])) - u256( int(service["reserved_wei"]) ) if available < max_payout: raise gl.vm.UserError( ERROR_EXPECTED + " provider collateral is fully reserved" ) service["reserved_wei"] = int( u256(int(service["reserved_wei"])) + max_payout ) service["provider_revenue_wei"] = int( u256(int(service["provider_revenue_wei"])) + price ) self.services[str(service["service_id"])] = json.dumps( service, sort_keys=True ) subscription["reserve_backed"] = True subscription["window_index"] = int(subscription.get("window_index", 1)) + 1 self.subscriptions[str(subscription_id)] = json.dumps( subscription, sort_keys=True ) @gl.public.view def get_service(self, service_id: str) -> str: return self.services.get(str(service_id), "") @gl.public.view def get_services(self) -> str: result = [] for service_id in self.service_order: result.append(json.loads(self.services[service_id])) return json.dumps(result, sort_keys=True) @gl.public.view def get_subscription(self, subscription_id: str) -> str: return self.subscriptions.get(str(subscription_id), "") @gl.public.view def get_subscriptions(self) -> str: result = [] for subscription_id in self.subscription_order: result.append(json.loads(self.subscriptions[subscription_id])) return json.dumps(result, sort_keys=True) @gl.public.view def get_snapshot(self, snapshot_id: str) -> str: return self.snapshots.get(str(snapshot_id), "") @gl.public.view def get_snapshots(self) -> str: result = [] for snapshot_id in self.snapshot_order: result.append(json.loads(self.snapshots[snapshot_id])) return json.dumps(result, sort_keys=True) @gl.public.view def get_attestation(self, attestation_id: str) -> str: return self.attestations.get(str(attestation_id).strip(), "") @gl.public.view def get_attestations(self) -> str: result = [] for attestation_id in self.attestation_order: result.append(json.loads(self.attestations[attestation_id])) return json.dumps(result, sort_keys=True) @gl.public.view def get_claim(self, claim_id: str) -> str: return self.claims.get(str(claim_id), "") @gl.public.view def get_claims(self) -> str: result = [] for claim_id in self.claim_order: result.append(json.loads(self.claims[claim_id])) return json.dumps(result, sort_keys=True) @gl.public.view def get_credit(self, credit_id: str) -> str: return self.credits.get(str(credit_id).strip(), "") @gl.public.view def get_credits(self) -> str: result = [] for credit_id in self.credit_order: result.append(json.loads(self.credits[credit_id])) return json.dumps(result, sort_keys=True) @gl.public.view def get_counts(self) -> str: return json.dumps( { "services": int(self.service_seq), "subscriptions": int(self.subscription_seq), "attestations": int(self.attestation_seq), "snapshots": int(self.snapshot_seq), "claims": int(self.claim_seq), "credits": int(self.credit_seq), }, sort_keys=True, )