# { "Depends": "py-genlayer:1jb45aa8ynh2a9c9xn3b7qqh8sm5q93hwfp7jqmwsfhh8jpz09h6" } from genlayer import * from dataclasses import dataclass from datetime import datetime, timezone import json import typing an = 400 at = 200 aq = 400 R = 70 U = 30 * 86400 F = 1 I = 365 * 86400 u = ("RECOMMEND", "RECOMMEND_OR_CAUTION") L = "[EXPECTED]" E = "QUEUED" V = "RELEASED" S = "CANCELLED" def ax(s: str) -> str: return " ".join(str(s).split()) def af(s: str, n: int) -> str: s = str(s) return s[:n] if len(s) > n else s def ag(s: typing.Any, n: int) -> str: M = [] for ch in str(s): M.append(ch if 32 <= ord(ch) < 127 else " ") return ax(ax("".join(M))[:n]) def h(v: typing.Any, au: int) -> int: if isinstance(v, bool) or not isinstance(v, int): return au return int(v) def ah(t: str, j: str) -> bool: if str(j) == "RECOMMEND": return True return str(t) == "RECOMMEND_OR_CAUTION" and str(j) == "CAUTION" @gl.contract_interface class IVoteGuard: class View: def get_risk_summary(self, d: str) -> typing.Any: ... def require_recommended(self, q: int) -> typing.Any: ... def get_config(self) -> typing.Any: ... class Write: pass @gl.evm.contract_interface class _Payee: class View: pass class Write: pass @allow_storage @dataclass class Payout: f: u32 d: str ai: str C: Address w: u256 T: Address D: str N: u64 K: u64 k: Address t: str b: u32 g: u64 j: str o: u32 q: u32 O: str class GovernanceConsumer(gl.Contract): Q: Address k: Address t: str b: u32 g: u64 r: DynArray[Payout] a: u256 z: TreeMap[Address, u256] c: u256 A: u256 l: u256 y: u256 def __init__(self, k: str): self.Q = gl.message.sender_address self.k = Address(str(k)) self.t = u[0] self.b = u32(R) self.g = u64(U) self.a = u256(0) self.c = u256(0) self.A = u256(0) self.l = u256(0) self.y = u256(0) def Y(self) -> int: return int(datetime.now(timezone.utc).timestamp()) def J(self) -> None: if gl.message.sender_address != self.Q: raise gl.vm.UserError(L + " owner only") def ay(self, ab: Address, e: int) -> None: if e <= 0: return self.z[ab] = u256(int(self.z.get(ab) or 0) + e) self.c = u256(int(self.c) + e) def B(self, O: str) -> dict: self.ay(gl.message.sender_address, int(gl.message.value)) return {"status": "REJECTED", "reason": af(O, 200), "refund_wei": int(gl.message.value)} def ac(self) -> int: Z = (int(gl.contract_balance) - int(self.a) - int(self.c)) return Z if Z > 0 else 0 def aj(self, k: Address, ao: str) -> dict: try: aC = IVoteGuard(k).view().get_risk_summary(str(ao)) except Exception: return {"known": False, "verdict": "UNREACHABLE"} if not isinstance(aC, dict): return {"known": False, "verdict": "UNREADABLE"} return aC def ad(self, p: Payout, now: int) -> dict: return { "payout_id": int(p.f), "proposal_url": str(p.d), "memo": str(p.ai), "recipient": str(p.C.as_hex), "amount_wei": int(p.w), "proposer": str(p.T.as_hex), "status": str(p.D), "queued_at": int(p.N), "settled_at": int(p.K), "age_seconds": now - int(p.N), "terms": {"oracle": str(p.k.as_hex), "mode": str(p.t), "min_score": int(p.b), "max_age_seconds": int(p.g)}, "verdict": str(p.j), "score": int(p.o), "assessment_id": int(p.q), "reason": str(p.O), } def av(self, f: int) -> typing.Any: i = h(f, -1) if i < 0 or i >= len(self.r): return None return self.r[i] @gl.public.write.payable def fund(self) -> typing.Any: return {"status": "OK", "added_wei": int(gl.message.value), "balance_wei": int(gl.contract_balance), "uncommitted_wei": self.ac()} @gl.public.view def preflight(self, d: str) -> typing.Any: s = self.aj(self.k, d) now = self.Y() ae = bool(s.get("known")) j = str(s.get("verdict", "UNKNOWN")) o = h(s.get("score"), 0) P = h(s.get("age_seconds"), 0) G = [] if not ae: G.append("no assessment on record") else: if not ah(str(self.t), j): G.append("verdict is " + j + ", this treasury " "requires " + str(self.t)) if o < int(self.b): G.append("score " + str(o) + " is below the required " + str(int(self.b))) if P > int(self.g): G.append("assessment is " + str(P) + "s old, older " "than the " + str(int(self.g)) + "s limit") return { "would_release": len(G) == 0, "proposal_url": af(str(d), an), "known": ae, "verdict": j, "score": o, "age_seconds": P, "worst_dimension": str(s.get("worst_dimension", "")), "worst_label": str(s.get("worst_label", "")), "flags": s.get("flags") if isinstance(s.get("flags"), list) else [], "title": str(s.get("title", "")), "dao": str(s.get("dao", "")), "blockers": G, "terms": {"oracle": str(self.k.as_hex), "mode": str(self.t), "min_score": int(self.b), "max_age_seconds": int(self.g)}, } @gl.public.view def strict_release(self, q: int) -> typing.Any: return IVoteGuard(self.k).view().require_recommended( h(q, -1)) @gl.public.write.payable def queue_payout(self, d: str, ai: str, C: str, w: int) -> typing.Any: value = int(gl.message.value) now = self.Y() ao = ag(d, an) if ao == "" or ao.find(" ") >= 0: return self.B("proposal url is empty or malformed") if not ao.lower().startswith("https://"): return self.B("proposal url must start with https://") try: to = Address(str(C)) except Exception: return self.B("recipient is not an address") if to == Address("0x" + "0" * 40): return self.B("recipient cannot be the zero address") e = h(w, -1) if e <= 0: return self.B("amount must be positive") if len(self.r) >= aq: return self.B("queue is full") if e > int(gl.contract_balance) - int(self.a) - int(self.c): return self.B( "treasury holds " + str(int(gl.contract_balance)) + " wei, of which " + str(int(self.a)) + " is already committed and " + str(int(self.c)) + " is owed as refunds") p = self.r.append_new_get() p.f = u32(len(self.r) - 1) p.d = ao p.ai = ag(ai, at) p.C = to p.w = u256(e) p.T = gl.message.sender_address p.D = E p.N = u64(now) p.k = self.k p.t = str(self.t) p.b = u32(int(self.b)) p.g = u64(int(self.g)) self.a = u256(int(self.a) + e) self.A = u256(int(self.A) + 1) M = self.ad(p, now) M["status_code"] = "OK" M["uncommitted_wei"] = self.ac() return M @gl.public.write def release(self, f: int) -> typing.Any: p = self.av(f) if p is None: raise gl.vm.UserError(L + " no such payout") if str(p.D) != E: raise gl.vm.UserError(L + " payout is already " + str(p.D).lower()) s = self.aj(p.k, str(p.d)) now = self.Y() ae = bool(s.get("known")) j = str(s.get("verdict", "UNKNOWN")) o = h(s.get("score"), 0) P = h(s.get("age_seconds"), 0) H = "" if not ae: H = "no assessment on record for that proposal" elif not ah(str(p.t), j): H = ("verdict is " + j + "; this payout requires " + str(p.t)) elif o < int(p.b): H = ("score " + str(o) + " is below the required " + str(int(p.b))) elif P > int(p.g): H = ("assessment is " + str(P) + "s old, older than the " + str(int(p.g)) + "s this payout allows") if H != "": self.y = u256(int(self.y) + 1) raise gl.vm.UserError( L + " refused: " + H + " (" + af(str(p.ai), 60) + ")") e = int(p.w) p.D = V p.K = u64(now) p.j = j p.o = u32(o) p.q = u32(h(s.get("assessment_id"), 0)) p.O = ag(s.get("title"), 120) self.a = u256(int(self.a) - e) self.l = u256(int(self.l) + e) _Payee(p.C).emit(value=u256(e)) M = self.ad(p, now) M["status_code"] = "RELEASED" return M @gl.public.write def cancel_payout(self, f: int) -> typing.Any: p = self.av(f) if p is None: raise gl.vm.UserError(L + " no such payout") if str(p.D) != E: raise gl.vm.UserError(L + " payout is already " + str(p.D).lower()) ak = gl.message.sender_address if ak != p.T and ak != self.Q: raise gl.vm.UserError(L + " only the queuer or the owner may " "cancel") p.D = S p.K = u64(self.Y()) p.O = "cancelled by " + str(ak.as_hex) self.a = u256(int(self.a) - int(p.w)) return {"status": "OK", "payout_id": int(p.f), "uncommitted_wei": self.ac()} @gl.public.write def claim_refund(self) -> typing.Any: ab = gl.message.sender_address e = int(self.z.get(ab) or 0) if e <= 0: return {"status": "NOTHING_OWED", "refund_wei": 0} self.z[ab] = u256(0) self.c = u256(int(self.c) - e) _Payee(ab).emit(value=u256(e)) return {"status": "OK", "refund_wei": e} @gl.public.view def get_payout(self, f: int) -> typing.Any: p = self.av(f) if p is None: return {"found": False, "payout_id": h(f, -1)} M = self.ad(p, self.Y()) M["found"] = True return M @gl.public.view def get_payouts(self, az: int, aB: int) -> typing.Any: W = h(az, 0) if W < 0: W = 0 n = h(aB, 0) if n <= 0 or n > 100: n = 100 now = self.Y() aw = [] for i in range(W, min(W + n, len(self.r))): aw.append(self.ad(self.r[i], now)) return {"total": len(self.r), "offset": W, "returned": len(aw), "payouts": aw} @gl.public.view def get_terms(self) -> typing.Any: al = {} try: ap = IVoteGuard(self.k).view().get_config() if isinstance(ap, dict): al = {"rubric_version": ap.get("rubric_version"), "verdicts": ap.get("verdicts"), "verdict_thresholds": ap.get("verdict_thresholds"), "fee_wei": ap.get("fee_wei"), "paused": ap.get("paused")} except Exception: al = {"error": "oracle unreachable"} return { "owner": str(self.Q.as_hex), "oracle": str(self.k.as_hex), "oracle_is_immutable": True, "mode": str(self.t), "accepted_modes": list(u), "min_score": int(self.b), "max_age_seconds": int(self.g), "committed_wei": int(self.a), "refunds_owed_wei": int(self.c), "queued": len(self.r), "total_queued": int(self.A), "total_released_wei": int(self.l), "total_refused": int(self.y), "oracle_rubric": al, "note": "terms are snapshotted into each payout when it is queued; " "set_terms moves the defaults for payouts queued after it", } @gl.public.view def refund_of(self, ab: str) -> typing.Any: aa = Address(str(ab)) return {"address": str(aa.as_hex), "refund_wei": int(self.z.get(aa) or 0)} @gl.public.write def set_terms(self, t: str, b: int, g: int) -> typing.Any: self.J() m = str(t) if m not in u: raise gl.vm.UserError(L + " mode must be one of " + ",".join(u)) o = h(b, -1) if o < F or o > 100: raise gl.vm.UserError(L + " min_score must be between " + str(F) + " and 100") P = h(g, -1) if P <= 0 or P > I: raise gl.vm.UserError(L + " max_age must be between 1 and " + str(I) + " seconds") aG = {"mode": str(self.t), "min_score": int(self.b), "max_age_seconds": int(self.g)} self.t = m self.b = u32(o) self.g = u64(P) return {"status": "OK", "was": aG, "now": {"mode": m, "min_score": o, "max_age_seconds": P}, "applies_to": "payouts queued after this call; the " + str(len(self.r)) + " already queued keep their own terms"} @gl.public.write def transfer_ownership(self, ar: str) -> typing.Any: self.J() aa = Address(str(ar)) if aa == Address("0x" + "0" * 40): raise gl.vm.UserError(L + " owner cannot be zero") self.Q = aa return {"status": "OK", "owner": str(aa.as_hex)} @gl.public.write def withdraw_uncommitted(self, e: int) -> typing.Any: self.J() am = h(e, -1) Z = self.ac() if am <= 0 or am > Z: raise gl.vm.UserError( L + " uncommitted balance is " + str(Z) + " wei (holds " + str(int(gl.contract_balance)) + ", committed " + str(int(self.a)) + ", owes " + str(int(self.c)) + ")") _Payee(self.Q).emit(value=u256(am)) return {"status": "OK", "withdrawn_wei": am, "uncommitted_wei": Z - am}