# { "Depends": "py-genlayer:1jb45aa8ynh2a9c9xn3b7qqh8sm5q93hwfp7jqmwsfhh8jpz09h6" } from genlayer import * from dataclasses import dataclass import json import hashlib CANARY_TOKEN = "CANARY_AGENT_GREEN_HYDROGEN_V1" ZERO_ADDRESS = "0x0000000000000000000000000000000000000000" # Time & Cooling-Off Parameters SECONDS_PER_BLOCK = 3 COOLING_OFF_BLOCKS = 24 COOLING_OFF_SECONDS = bigint(COOLING_OFF_BLOCKS * SECONDS_PER_BLOCK) # 72 seconds DEFAULT_SUBSIDY_EXPIRY_SECONDS = bigint(86400 * 30) # 30 days coverage # Lifecycle Statuses STATUS_POLICY_OPEN = u8(0) # Climate Fund deposited subsidy escrow, awaiting production telemetry STATUS_TELEMETRY_SUBMITTED = u8(1) # Electrolyzer plant submitted gas chromatography and renewable certificates STATUS_AWAITING_PAYOUT = u8(2) # AI Hydrogen Jury rendered verdict, cooling-off active STATUS_SETTLED_CERTIFIED_FULL = u8(3)# ISO 14687 compliant green hydrogen confirmed, 100% subsidy released STATUS_SETTLED_PARTIAL = u8(4) # Carbon intensity breach / minor grid mixing, 50% penalty STATUS_SETTLED_REJECTED = u8(5) # Gray hydrogen detected / purity failure, 100% refunded to fund STATUS_DISPUTED = u8(6) # Under appellate review with 10% staked bond STATUS_EXPIRED_RECLAIMED = u8(7) # Policy expired with zero telemetry, reclaimed by fund STATUS_UNRESOLVED = u8(8) # Telemetry fetch, parse, or canary failure (protected non-settling state) class ContractError(Exception): """Domain-specific error for AgentGreenHydrogen protocol.""" pass def _to_address(val) -> Address: """Converts any Address, int, or str into a valid GenLayer Address instance.""" if isinstance(val, Address): return val if isinstance(val, int): hex_str = f"0x{val:040x}" return Address(hex_str) if isinstance(val, str): clean = val.strip() if clean.startswith("0x") or clean.startswith("0X"): return Address(clean.lower()) try: int_val = int(clean) return Address(f"0x{int_val:040x}") except ValueError: return Address(clean.lower()) if isinstance(val, (bytes, bytearray)): return Address("0x" + val.hex()) return Address(str(val)) def _addr_str(addr) -> str: """Safely format an Address or address-like value into a lowercase hex string.""" if addr is None: return ZERO_ADDRESS if isinstance(addr, int): return f"0x{addr:040x}".lower() try: return addr.as_hex.lower() except Exception: return str(addr).lower() def _get_sender() -> Address: """Safely obtain transaction sender across GenVM runtime versions.""" try: raw = gl.message.sender_address return _to_address(raw) except Exception: try: raw = gl.message.sender return _to_address(raw) except Exception: raise ContractError("Cannot resolve sender address.") ACCREDITED_HYDROGEN_AUDITORS = [ "tuvsud.com", "dnv.com", "bureauveritas.com", "nrel.gov", "certifhy.eu", "energy.gov", "fraunhofer.de", "iso.org", ] def _is_accredited_hydrogen_auditor(url: str) -> bool: """Verifies that gas chromatography telemetry originates strictly from accredited testing agencies.""" clean = str(url).strip().lower() if not (clean.startswith("http://") or clean.startswith("https://")): return False try: no_proto = clean.split("://", 1)[1] host = no_proto.split("/", 1)[0].split(":", 1)[0] for domain in ACCREDITED_HYDROGEN_AUDITORS: if host == domain or host.endswith("." + domain): return True except Exception: return False return False @allow_storage @dataclass class HydrogenPolicy: policy_id: str sponsor: Address # Climate Subsidy DAO / Clean Energy Fund electrolyzer_plant: Address # Green Hydrogen Production Facility dispute_initiator: Address # Appellant who staked bond escrow_amount: bigint # Locked subsidy allocation dispute_bond: bigint # 10% appeal stake facility_id: str # Production plant identifier / GPS coordinates min_purity_bps: u32 # Basis points of purity (e.g., 9997 = 99.97% ISO 14687) max_carbon_intensity_g_co2: u32# Max allowable g CO2e per kg H2 (e.g., 2000g = 2.0 kg CO2/kg H2) chromatography_feed_url: str # Accredited live lab report / gas chromatography telemetry evidence_hash: str # SHA-256 snapshot of laboratory telemetry payload status: u8 verdict: str # "PENDING", "GREEN_HYDROGEN_CERTIFIED_FULL", "PARTIAL_CARBON_INTENSITY_PENALTY", "GRAY_HYDROGEN_REJECTED", "DISPUTED", "UNRESOLVED_*" original_verdict: str # Preserved pre-appeal verdict reason: str confidence: u8 measured_purity_bps: u32 # Measured purity in basis points (e.g. 9998 = 99.98%) measured_carbon_intensity_g: u32 # Measured carbon intensity in grams CO2e/kg H2 created_at_time: bigint # Trusted Unix timestamp when policy was created expires_at_time: bigint # Trusted Unix timestamp when coverage concludes audit_completed_time: bigint # Trusted Unix timestamp when AI jury rendered verdict supplemental_evidence_url: str # Bound appellate evidence URL appeal_prior_verdict: str # Verdict recorded before appeal was filed class Contract(gl.Contract): """ AgentGreenHydrogen: Autonomous Green Hydrogen Purity & Renewable Electrolysis Subsidy Escrow Target Network: GenLayer StudioNet (Chain ID: 61999) """ policies: TreeMap[str, HydrogenPolicy] policy_ids: DynArray[str] total_subsidies_locked: bigint total_policies_settled: u32 policy_counter: u64 owner: Address claimable_balances: TreeMap[str, bigint] def __init__(self, initial_owner: Address = Address(ZERO_ADDRESS)): target_owner = _to_address(initial_owner) if initial_owner is not None else Address(ZERO_ADDRESS) if _addr_str(target_owner) == ZERO_ADDRESS: try: target_owner = _to_address(_get_sender()) except Exception: pass self.owner = target_owner self.total_subsidies_locked = bigint(0) self.total_policies_settled = u32(0) self.policy_counter = u64(0) def _resolve_pid(self, policy_id) -> str: return str(policy_id).strip() def _ensure_owner(self) -> None: if _addr_str(self.owner) == ZERO_ADDRESS: self.owner = _to_address(_get_sender()) def _safe_transfer(self, recipient: Address, amount: bigint) -> None: """Transfers native GEN tokens with Pull-Payment fallback to eliminate lost funds.""" if amount <= bigint(0): return r_addr = _to_address(recipient) rec_str = _addr_str(r_addr) try: gl.get_contract_at(r_addr).emit_transfer(value=u256(int(amount))) except Exception: curr = self.claimable_balances.get(rec_str, bigint(0)) self.claimable_balances[rec_str] = curr + amount def _get_current_timestamp(self) -> bigint: """Derives trusted Unix timestamp strictly from consensus context (gl.message_raw['datetime']).""" import calendar from datetime import datetime, timezone try: if hasattr(gl, "message_raw") and isinstance(gl.message_raw, dict): raw_val = gl.message_raw.get("datetime", "") if raw_val: dt_str = str(raw_val).strip().replace("Z", "+00:00") dt = datetime.fromisoformat(dt_str) if dt.tzinfo is None: dt = dt.replace(tzinfo=timezone.utc) ts = calendar.timegm(dt.utctimetuple()) if ts > 0: return bigint(ts) except Exception: pass return bigint(1760000000) # ── Public Write Methods (Strict Role Enforcement) ──────────────── @gl.public.write.payable def create_subsidy_policy( self, electrolyzer_plant: Address, facility_id: str, min_purity_bps: u32, max_carbon_intensity_g_co2: u32, duration_blocks: u32 ) -> str: """Role: Climate Subsidy Fund deposits subsidy allocation for accredited hydrogen electrolysis.""" self._ensure_owner() escrow = bigint(gl.message.value) if escrow <= bigint(0): raise ContractError("Subsidy escrow deposit must be greater than 0 GEN.") clean_fac = str(facility_id).strip() if len(clean_fac) < 4: raise ContractError("Valid hydrogen production facility ID required (>= 4 chars).") p_addr = _to_address(electrolyzer_plant) if _addr_str(p_addr) == ZERO_ADDRESS: raise ContractError("Valid electrolyzer plant address required.") s_addr = _to_address(_get_sender()) if _addr_str(s_addr) == _addr_str(p_addr): raise ContractError("Role Violation: Sponsor fund cannot designate themselves as the electrolyzer plant.") dur_secs = bigint(int(duration_blocks) * 3 if int(duration_blocks) > 0 else int(DEFAULT_SUBSIDY_EXPIRY_SECONDS)) self.policy_counter = self.policy_counter + u64(1) policy_id = str(self.policy_counter) current_time = self._get_current_timestamp() expires_at = current_time + dur_secs empty_addr = Address(ZERO_ADDRESS) new_policy = HydrogenPolicy( policy_id=policy_id, sponsor=s_addr, electrolyzer_plant=p_addr, dispute_initiator=empty_addr, escrow_amount=escrow, dispute_bond=bigint(0), facility_id=clean_fac, min_purity_bps=u32(int(min_purity_bps)), max_carbon_intensity_g_co2=u32(int(max_carbon_intensity_g_co2)), chromatography_feed_url="", evidence_hash="", status=STATUS_POLICY_OPEN, verdict="PENDING", original_verdict="PENDING", reason="Subsidy escrow active. Monitored for ISO 14687 hydrogen purity and 100% renewable energy attribution.", confidence=u8(0), measured_purity_bps=u32(0), measured_carbon_intensity_g=u32(0), created_at_time=current_time, expires_at_time=expires_at, audit_completed_time=bigint(0), supplemental_evidence_url="", appeal_prior_verdict="", ) self.policies[policy_id] = new_policy self.policy_ids.append(policy_id) self.total_subsidies_locked = self.total_subsidies_locked + escrow return policy_id @gl.public.write def submit_production_telemetry( self, policy_id: str, chromatography_feed_url: str ) -> None: """Role: Electrolyzer facility submits accredited gas chromatography and hourly renewable matching data.""" self._ensure_owner() pid = self._resolve_pid(policy_id) if pid not in self.policies: raise ContractError(f"Policy {pid} does not exist.") p = self.policies[pid] if p.status not in (STATUS_POLICY_OPEN, STATUS_UNRESOLVED): raise ContractError("Policy is not open for production telemetry submission.") sender = _get_sender() if _addr_str(sender) != _addr_str(p.electrolyzer_plant): raise ContractError("Role Violation: Only registered electrolyzer plant can submit production telemetry.") clean_url = str(chromatography_feed_url).strip() if not _is_accredited_hydrogen_auditor(clean_url): raise ContractError("Telemetry unauthenticated: Must originate from an accredited testing agency (TÜV SÜD, DNV, Bureau Veritas, NREL, CertifHy, Fraunhofer).") current_time = self._get_current_timestamp() if current_time > p.expires_at_time: raise ContractError("Coverage duration for this subsidy policy has expired.") p.chromatography_feed_url = clean_url p.status = STATUS_TELEMETRY_SUBMITTED p.reason = "Gas chromatography and renewable matching telemetry submitted. Ready for on-chain AI Tribunal verification." @gl.public.write def adjudicate_hydrogen_quality(self, policy_id: str) -> None: """ Role: Stakeholders or Owner invoke AI Hydrogen Tribunal. Non-deterministic Consensus: 1. gl.nondet.web.render pulls live chromatography and carbon intensity telemetry. 2. gl.nondet.exec_prompt evaluates ISO 14687 hydrogen purity & hourly renewable matching. """ self._ensure_owner() pid = self._resolve_pid(policy_id) if pid not in self.policies: raise ContractError(f"Policy {pid} does not exist.") p = self.policies[pid] if p.status not in (STATUS_TELEMETRY_SUBMITTED, STATUS_UNRESOLVED): raise ContractError("Policy is not in a valid state for AI quality adjudication.") sender = _get_sender() sender_str = _addr_str(sender) if ( sender_str != _addr_str(p.sponsor) and sender_str != _addr_str(p.electrolyzer_plant) and sender_str != _addr_str(self.owner) ): raise ContractError("Permission Denied: Only sponsor, electrolyzer plant, or owner can trigger adjudication.") # Step 1: Web render from accredited laboratory feed chromatography_url = p.chromatography_feed_url web_content = "" fetch_error = False try: web_content = gl.nondet.web.render( chromatography_url, mode="text", wait_until="networkidle", timeout=15000 ) except Exception: fetch_error = True if fetch_error or not web_content or len(web_content.strip()) < 10: p.status = STATUS_UNRESOLVED p.verdict = "UNRESOLVED_FETCH_ERROR" p.reason = "Accredited telemetry feed unreachable or returned empty payload. Policy kept intact for retry." return evidence_hash = hashlib.sha256(web_content.encode("utf-8")).hexdigest() p.evidence_hash = evidence_hash # Step 2: AI Hydrogen Jury Consensus Evaluation eval_prompt = f""" You are the Autonomous Green Hydrogen Purity & Renewable Electrolysis Auditor for GenLayer. Evaluate the following gas chromatography and renewable electricity telemetry against ISO 14687 standards and contract parameters. Target Policy Requirements: - Facility ID: {p.facility_id} - Min Allowable Hydrogen Purity (bps): {int(p.min_purity_bps)} (e.g. 9997 = 99.97% ISO 14687 Fuel Cell Grade) - Max Allowable Carbon Intensity: {int(p.max_carbon_intensity_g_co2)} g CO2e / kg H2 (e.g. 2000 g = 2.0 kg CO2/kg H2 threshold) Accredited Laboratory Telemetry Content: {web_content[:4000]} Adjudication Rules: 1. "GREEN_HYDROGEN_CERTIFIED_FULL": Hydrogen Purity >= {int(p.min_purity_bps)} bps AND Carbon Intensity <= {int(p.max_carbon_intensity_g_co2)} g CO2e/kg H2 AND 100% matched renewable power. 2. "PARTIAL_CARBON_INTENSITY_PENALTY": Hydrogen Purity >= {int(p.min_purity_bps)} bps BUT Carbon Intensity exceeds threshold by up to 25% or grid mix exceeds 10%. 3. "GRAY_HYDROGEN_REJECTED": Hydrogen Purity < {int(p.min_purity_bps)} bps OR Carbon Intensity indicates fossil gas steam methane reforming (SMR) or persistent grid fossil reliance. CANARY REQUIREMENT: You MUST include "{CANARY_TOKEN}" in the "canary" JSON field. Output strictly valid JSON matching this schema: {{ "canary": "{CANARY_TOKEN}", "verdict": "GREEN_HYDROGEN_CERTIFIED_FULL" | "PARTIAL_CARBON_INTENSITY_PENALTY" | "GRAY_HYDROGEN_REJECTED", "confidence": 0-100, "measured_purity_bps": integer, "measured_carbon_intensity_g": integer, "reason": "Clear explanation of chromatography analysis and carbon footprint" }} """ raw_decision = "" try: raw_decision = gl.nondet.exec_prompt(eval_prompt) except Exception: p.status = STATUS_UNRESOLVED p.verdict = "UNRESOLVED_AI_ERROR" p.reason = "AI Hydrogen Tribunal execution failed. Non-fatal state preserved." return cleaned_json = raw_decision.strip() if cleaned_json.startswith("```json"): cleaned_json = cleaned_json[7:] elif cleaned_json.startswith("```"): cleaned_json = cleaned_json[3:] if cleaned_json.endswith("```"): cleaned_json = cleaned_json[:-3] cleaned_json = cleaned_json.strip() try: decision = json.loads(cleaned_json) except Exception: p.status = STATUS_UNRESOLVED p.verdict = "UNRESOLVED_PARSE_ERROR" p.reason = "AI Jury output invalid JSON. Telemetry intact for retry." return # Canary verification if decision.get("canary") != CANARY_TOKEN: p.status = STATUS_UNRESOLVED p.verdict = "UNRESOLVED_CANARY_FAIL" p.reason = "Consensus canary token missing or invalid. Potential adversarial tampering." return verdict = str(decision.get("verdict", "")).strip() conf = int(decision.get("confidence", 0)) reason = str(decision.get("reason", "Audited by GenLayer AI Tribunal."))[:500] purity_bps = int(decision.get("measured_purity_bps", 0)) carbon_g = int(decision.get("measured_carbon_intensity_g", 0)) if verdict not in ("GREEN_HYDROGEN_CERTIFIED_FULL", "PARTIAL_CARBON_INTENSITY_PENALTY", "GRAY_HYDROGEN_REJECTED"): p.status = STATUS_UNRESOLVED p.verdict = "UNRESOLVED_INVALID_VERDICT" p.reason = f"Unrecognized tribunal verdict: {verdict}" return p.verdict = verdict p.original_verdict = verdict p.confidence = u8(conf if 0 <= conf <= 100 else 80) p.reason = reason p.measured_purity_bps = u32(purity_bps if purity_bps > 0 else 0) p.measured_carbon_intensity_g = u32(carbon_g if carbon_g > 0 else 0) p.audit_completed_time = self._get_current_timestamp() p.status = STATUS_AWAITING_PAYOUT @gl.public.write.payable def file_dispute( self, policy_id: str, counter_evidence_url: str ) -> None: """ Role: Sponsor or Electrolyzer Facility files appeal within cooling-off window. Must stake 10% dispute bond to deter frivolous appellate disruption. """ self._ensure_owner() pid = self._resolve_pid(policy_id) if pid not in self.policies: raise ContractError(f"Policy {pid} does not exist.") p = self.policies[pid] if p.status != STATUS_AWAITING_PAYOUT: raise ContractError("Dispute can only be lodged during cooling-off window (STATUS_AWAITING_PAYOUT).") current_time = self._get_current_timestamp() if current_time > (p.audit_completed_time + COOLING_OFF_SECONDS): raise ContractError("Cooling-off challenge window has elapsed. Settlement can now be finalized.") sender = _get_sender() sender_str = _addr_str(sender) if sender_str != _addr_str(p.sponsor) and sender_str != _addr_str(p.electrolyzer_plant): raise ContractError("Permission Denied: Only sponsor or electrolyzer plant can challenge the verdict.") # Require 10% dispute bond required_bond = bigint(int(p.escrow_amount) // 10) paid_bond = bigint(gl.message.value) if paid_bond < required_bond: raise ContractError(f"Insufficient dispute bond. Required 10%: {required_bond} wei.") clean_url = str(counter_evidence_url).strip() if not _is_accredited_hydrogen_auditor(clean_url): raise ContractError("Counter-evidence must originate from an accredited testing agency.") p.dispute_initiator = sender p.dispute_bond = paid_bond p.supplemental_evidence_url = clean_url p.appeal_prior_verdict = p.verdict p.status = STATUS_DISPUTED p.reason = f"Appellate challenge filed by {sender_str}. Staked {paid_bond} wei bond. Pending Appellate Tribunal." @gl.public.write def adjudicate_appeal(self, policy_id: str) -> None: """ Role: AI High Tribunal adjudicates appellate challenge against original verdict and counter-evidence. Bond Distribution: - If appellant wins (verdict overturned): 100% bond returned to appellant. - If appeal rejected: Bond forfeited 100% to opponent. """ self._ensure_owner() pid = self._resolve_pid(policy_id) if pid not in self.policies: raise ContractError(f"Policy {pid} does not exist.") p = self.policies[pid] if p.status != STATUS_DISPUTED: raise ContractError("Policy is not in STATUS_DISPUTED.") sender = _get_sender() sender_str = _addr_str(sender) if ( sender_str != _addr_str(p.sponsor) and sender_str != _addr_str(p.electrolyzer_plant) and sender_str != _addr_str(self.owner) ): raise ContractError("Permission Denied: Only sponsor, electrolyzer plant, or owner can adjudicate appeal.") # Render counter-evidence web_content = "" fetch_err = False try: web_content = gl.nondet.web.render( p.supplemental_evidence_url, mode="text", wait_until="networkidle", timeout=15000 ) except Exception: fetch_err = True if fetch_err or not web_content or len(web_content.strip()) < 10: p.status = STATUS_DISPUTED p.reason = "Supplemental evidence feed unreachable. Appeal remains pending." return appellate_prompt = f""" You are the High Court Appellate Tribunal for Green Hydrogen Certification on GenLayer. An appeal has been lodged challenging the initial verdict for electrolyzer facility {p.facility_id}. Prior Verdict: {p.appeal_prior_verdict} Initial Purity Requirement: {int(p.min_purity_bps)} bps Initial Carbon Cap: {int(p.max_carbon_intensity_g_co2)} g CO2e/kg H2 Supplemental Appellate Evidence from Accredited Auditor: {web_content[:4000]} Adjudication Objective: Determine whether the supplemental laboratory data conclusively upholds or overturns the prior verdict. CANARY REQUIREMENT: Include "{CANARY_TOKEN}" in "canary" JSON field. Output strictly valid JSON: {{ "canary": "{CANARY_TOKEN}", "appeal_upheld": true | false, "final_verdict": "GREEN_HYDROGEN_CERTIFIED_FULL" | "PARTIAL_CARBON_INTENSITY_PENALTY" | "GRAY_HYDROGEN_REJECTED", "measured_purity_bps": integer, "measured_carbon_intensity_g": integer, "confidence": 0-100, "rationale": "Detailed appellate justification" }} """ raw_res = "" try: raw_res = gl.nondet.exec_prompt(appellate_prompt) except Exception: return cleaned_json = raw_res.strip() if cleaned_json.startswith("```json"): cleaned_json = cleaned_json[7:] elif cleaned_json.startswith("```"): cleaned_json = cleaned_json[3:] if cleaned_json.endswith("```"): cleaned_json = cleaned_json[:-3] cleaned_json = cleaned_json.strip() try: res = json.loads(cleaned_json) except Exception: return if res.get("canary") != CANARY_TOKEN: return final_verdict = str(res.get("final_verdict", "")).strip() if final_verdict not in ("GREEN_HYDROGEN_CERTIFIED_FULL", "PARTIAL_CARBON_INTENSITY_PENALTY", "GRAY_HYDROGEN_REJECTED"): return appeal_upheld = bool(res.get("appeal_upheld", False)) rationale = str(res.get("rationale", "High Tribunal appellate ruling completed."))[:500] purity = int(res.get("measured_purity_bps", 0)) carbon = int(res.get("measured_carbon_intensity_g", 0)) conf = int(res.get("confidence", 90)) p.verdict = final_verdict p.reason = rationale p.measured_purity_bps = u32(purity if purity > 0 else 0) p.measured_carbon_intensity_g = u32(carbon if carbon > 0 else 0) p.confidence = u8(conf if 0 <= conf <= 100 else 90) # Distribute dispute bond and payout according to appellate verdict bond = p.dispute_bond appellant = p.dispute_initiator p.dispute_bond = bigint(0) # Determine opponent opponent = p.electrolyzer_plant if _addr_str(appellant) == _addr_str(p.sponsor) else p.sponsor if appeal_upheld: # Appellant prevailed: return bond in full to appellant self._safe_transfer(appellant, bond) else: # Appeal denied: bond forfeited to opponent self._safe_transfer(opponent, bond) # Finalize settlement directly post-appeal escrow = p.escrow_amount p.escrow_amount = bigint(0) if final_verdict == "GREEN_HYDROGEN_CERTIFIED_FULL": self._safe_transfer(p.electrolyzer_plant, escrow) p.status = STATUS_SETTLED_CERTIFIED_FULL elif final_verdict == "PARTIAL_CARBON_INTENSITY_PENALTY": payout = bigint(int(escrow) // 2) refund = escrow - payout self._safe_transfer(p.electrolyzer_plant, payout) self._safe_transfer(p.sponsor, refund) p.status = STATUS_SETTLED_PARTIAL else: self._safe_transfer(p.sponsor, escrow) p.status = STATUS_SETTLED_REJECTED self.total_policies_settled = self.total_policies_settled + u32(1) @gl.public.write def finalize_settlement(self, policy_id: str) -> None: """Role: Stakeholders or owner executes uncontested settlement strictly after cooling-off window.""" self._ensure_owner() pid = self._resolve_pid(policy_id) if pid not in self.policies: raise ContractError(f"Policy {pid} does not exist.") p = self.policies[pid] if p.status != STATUS_AWAITING_PAYOUT: raise ContractError("Policy is not in STATUS_AWAITING_PAYOUT state.") current_time = self._get_current_timestamp() if current_time <= (p.audit_completed_time + COOLING_OFF_SECONDS): remaining = (p.audit_completed_time + COOLING_OFF_SECONDS) - current_time raise ContractError(f"Cooling-off period active. Payout locked for another {remaining} seconds.") sender = _get_sender() sender_str = _addr_str(sender) if ( sender_str != _addr_str(p.sponsor) and sender_str != _addr_str(p.electrolyzer_plant) and sender_str != _addr_str(self.owner) ): raise ContractError("Permission Denied: Only stakeholders or owner can finalize payout.") escrow = p.escrow_amount p.escrow_amount = bigint(0) if p.verdict == "GREEN_HYDROGEN_CERTIFIED_FULL": # 100% subsidy released to clean hydrogen producer self._safe_transfer(p.electrolyzer_plant, escrow) p.status = STATUS_SETTLED_CERTIFIED_FULL p.reason = f"Uncontested ISO 14687 certification settled. 100% subsidy ({escrow} wei) transferred to electrolyzer plant." elif p.verdict == "PARTIAL_CARBON_INTENSITY_PENALTY": # 50% penalty for excess carbon intensity / grid reliance half = bigint(int(escrow) // 2) rem = escrow - half self._safe_transfer(p.electrolyzer_plant, half) self._safe_transfer(p.sponsor, rem) p.status = STATUS_SETTLED_PARTIAL p.reason = f"Partial carbon compliance settled. 50% penalty applied. {half} wei to plant, {rem} wei refunded to climate fund." elif p.verdict == "GRAY_HYDROGEN_REJECTED": # 100% refunded to climate fund due to gray hydrogen or purity violation self._safe_transfer(p.sponsor, escrow) p.status = STATUS_SETTLED_REJECTED p.reason = f"Gray hydrogen rejected. 100% subsidy escrow ({escrow} wei) clawed back to climate fund." else: # Fallback safety: return escrow to fund self._safe_transfer(p.sponsor, escrow) p.status = STATUS_SETTLED_REJECTED self.total_policies_settled = self.total_policies_settled + u32(1) @gl.public.write def reclaim_expired_subsidy(self, policy_id: str) -> None: """Role: Sponsor reclaims subsidy if plant failed to submit telemetry within the agreed coverage lifespan.""" self._ensure_owner() pid = self._resolve_pid(policy_id) if pid not in self.policies: raise ContractError(f"Policy {pid} does not exist.") p = self.policies[pid] if p.status not in (STATUS_POLICY_OPEN, STATUS_UNRESOLVED): raise ContractError("Subsidy policy cannot be reclaimed in its current lifecycle state.") current_time = self._get_current_timestamp() if current_time <= p.expires_at_time: raise ContractError("Policy coverage period is still active.") sender = _get_sender() if _addr_str(sender) != _addr_str(p.sponsor) and _addr_str(sender) != _addr_str(self.owner): raise ContractError("Permission Denied: Only sponsor fund or contract owner can reclaim expired subsidy.") escrow = p.escrow_amount p.escrow_amount = bigint(0) self._safe_transfer(p.sponsor, escrow) p.status = STATUS_EXPIRED_RECLAIMED p.reason = "Coverage expired with zero verified telemetry. Escrow reclaimed by sponsor fund." self.total_policies_settled = self.total_policies_settled + u32(1) @gl.public.write def claim_payout(self) -> None: """Pull-Payment pattern: Withdraw any accumulated claimable balances without risk of contract lock.""" self._ensure_owner() sender = _get_sender() sender_str = _addr_str(sender) bal = self.claimable_balances.get(sender_str, bigint(0)) if bal <= bigint(0): raise ContractError("No claimable balance available for this address.") self.claimable_balances[sender_str] = bigint(0) try: gl.get_contract_at(sender).emit_transfer(value=u256(int(bal))) except Exception: self.claimable_balances[sender_str] = bal raise ContractError("Withdrawal transfer failed. Balance remains securely claimable.") # ── Public View Methods (Deterministic Consensus) ──────────────── @gl.public.view def get_policy(self, policy_id: str) -> str: pid = self._resolve_pid(policy_id) if pid not in self.policies: return "{}" p = self.policies[pid] data = { "policy_id": p.policy_id, "sponsor": _addr_str(p.sponsor), "electrolyzer_plant": _addr_str(p.electrolyzer_plant), "dispute_initiator": _addr_str(p.dispute_initiator), "escrow_amount": str(p.escrow_amount), "dispute_bond": str(p.dispute_bond), "facility_id": p.facility_id, "min_purity_bps": int(p.min_purity_bps), "max_carbon_intensity_g_co2": int(p.max_carbon_intensity_g_co2), "chromatography_feed_url": p.chromatography_feed_url, "evidence_hash": p.evidence_hash, "status": int(p.status), "verdict": p.verdict, "original_verdict": p.original_verdict, "reason": p.reason, "confidence": int(p.confidence), "measured_purity_bps": int(p.measured_purity_bps), "measured_carbon_intensity_g": int(p.measured_carbon_intensity_g), "created_at_time": str(p.created_at_time), "expires_at_time": str(p.expires_at_time), "audit_completed_time": str(p.audit_completed_time), "supplemental_evidence_url": p.supplemental_evidence_url, "appeal_prior_verdict": p.appeal_prior_verdict, } return json.dumps(data) @gl.public.view def get_user_role(self, policy_id: str, account: Address) -> str: """Determines user's role: SPONSOR, ELECTROLYZER_PLANT, APPELLANT, OWNER, or OBSERVER.""" pid = self._resolve_pid(policy_id) acc_str = _addr_str(account) if pid not in self.policies: if acc_str == _addr_str(self.owner): return "OWNER" return "OBSERVER" p = self.policies[pid] if acc_str == _addr_str(p.sponsor): return "SPONSOR" if acc_str == _addr_str(p.electrolyzer_plant): return "ELECTROLYZER_PLANT" if acc_str == _addr_str(p.dispute_initiator) and acc_str != ZERO_ADDRESS: return "APPELLANT" if acc_str == _addr_str(self.owner): return "OWNER" return "OBSERVER" @gl.public.view def get_remaining_cooling_off_seconds(self, policy_id: str) -> i64: """Returns remaining cooling off challenge window in seconds (0 if expired or not in cooling off).""" pid = self._resolve_pid(policy_id) if pid not in self.policies: return i64(0) p = self.policies[pid] if p.status != STATUS_AWAITING_PAYOUT or p.audit_completed_time <= bigint(0): return i64(0) current_time = self._get_current_timestamp() expiry = p.audit_completed_time + COOLING_OFF_SECONDS remaining = int(expiry - current_time) return i64(remaining if remaining > 0 else 0) @gl.public.view def get_policy_count(self) -> u64: return u64(len(self.policy_ids)) @gl.public.view def get_all_policies(self) -> str: policies_list = [] for pid in self.policy_ids: if pid in self.policies: p = self.policies[pid] policies_list.append({ "policy_id": p.policy_id, "sponsor": _addr_str(p.sponsor), "electrolyzer_plant": _addr_str(p.electrolyzer_plant), "dispute_initiator": _addr_str(p.dispute_initiator), "escrow_amount": str(p.escrow_amount), "dispute_bond": str(p.dispute_bond), "facility_id": p.facility_id, "min_purity_bps": int(p.min_purity_bps), "max_carbon_intensity_g_co2": int(p.max_carbon_intensity_g_co2), "chromatography_feed_url": p.chromatography_feed_url, "evidence_hash": p.evidence_hash, "status": int(p.status), "verdict": p.verdict, "original_verdict": p.original_verdict, "reason": p.reason, "confidence": int(p.confidence), "measured_purity_bps": int(p.measured_purity_bps), "measured_carbon_intensity_g": int(p.measured_carbon_intensity_g), "created_at_time": str(p.created_at_time), "expires_at_time": str(p.expires_at_time), "audit_completed_time": str(p.audit_completed_time), }) return json.dumps(policies_list) @gl.public.view def get_claimable_balance(self, account: Address) -> str: return str(self.claimable_balances.get(_addr_str(account), bigint(0))) @gl.public.view def get_stats(self) -> str: data = { "total_policies": len(self.policy_ids), "total_subsidies_locked": str(self.total_subsidies_locked), "total_policies_settled": int(self.total_policies_settled), "owner": _addr_str(self.owner), } return json.dumps(data)