# v0.2.17 # { "Depends": "py-genlayer:1jb45aa8ynh2a9c9xn3b7qqh8sm5q93hwfp7jqmwsfhh8jpz09h6" } from genlayer import * import json class EmergencyGuard(gl.Contract): paused_protocols: TreeMap[str, bool] warnings: DynArray[str] emergency_log: DynArray[str] admin: Address global_pause: bool def __init__(self) -> None: self.admin = gl.message.sender_address self.paused_protocols = TreeMap() self.global_pause = False @gl.public.view def is_protocol_paused(self, protocol: str) -> bool: if self.global_pause: return True return self.paused_protocols.get(protocol, False) @gl.public.view def is_global_pause_active(self) -> bool: return self.global_pause @gl.public.view def get_admin(self) -> str: return str(self.admin) @gl.public.view def get_warnings(self, count: int) -> str: total = len(self.warnings) if count <= 0: return json.dumps([]) start = total - count if start < 0: start = 0 items = [] i = start while i < total: items.append(self.warnings[i]) i += 1 return json.dumps(items) @gl.public.view def get_emergency_log(self, count: int) -> str: total = len(self.emergency_log) if count <= 0: return json.dumps([]) start = total - count if start < 0: start = 0 items = [] i = start while i < total: items.append(self.emergency_log[i]) i += 1 return json.dumps(items) @gl.public.write def pause_protocol(self, protocol: str, reason: str) -> None: assert gl.message.sender_address == self.admin, "Only admin" self.paused_protocols[protocol] = True log_entry = json.dumps({ "action": "pause", "protocol": protocol, "reason": reason, "timestamp": str(gl.block.timestamp), "admin": str(gl.message.sender_address), }) self.emergency_log.append(log_entry) @gl.public.write def unpause_protocol(self, protocol: str, reason: str) -> None: assert gl.message.sender_address == self.admin, "Only admin" self.paused_protocols[protocol] = False log_entry = json.dumps({ "action": "unpause", "protocol": protocol, "reason": reason, "timestamp": str(gl.block.timestamp), "admin": str(gl.message.sender_address), }) self.emergency_log.append(log_entry) @gl.public.write def trigger_global_pause(self, reason: str) -> None: assert gl.message.sender_address == self.admin, "Only admin" self.global_pause = True log_entry = json.dumps({ "action": "global_pause", "reason": reason, "timestamp": str(gl.block.timestamp), "admin": str(gl.message.sender_address), }) self.emergency_log.append(log_entry) @gl.public.write def lift_global_pause(self, reason: str) -> None: assert gl.message.sender_address == self.admin, "Only admin" self.global_pause = False log_entry = json.dumps({ "action": "lift_global_pause", "reason": reason, "timestamp": str(gl.block.timestamp), "admin": str(gl.message.sender_address), }) self.emergency_log.append(log_entry) @gl.public.write def ai_evaluate_emergency( self, protocol: str, incident_description: str, ) -> str: prompt = f""" You are SentraIQ's Emergency Response AI. Evaluate the following DeFi incident and determine if emergency protective measures are required. Protocol: {protocol} Incident: {incident_description} Assess: 1. Severity level 2. Immediate risk to user funds 3. Whether protocol should be paused immediately 4. Recommended response actions Respond ONLY with JSON: {{ "severity": "low|medium|high|critical", "immediate_risk": true/false, "recommend_pause": true/false, "actions": ["", ""], "reasoning": "", "confidence": <0-100> }} """ raw = gl.exec_prompt(prompt) try: result = json.loads(raw.strip()) except Exception: result = { "severity": "high", "immediate_risk": True, "recommend_pause": True, "actions": [ "Pause protocol immediately", "Alert all depositors", ], "reasoning": "Incident requires immediate review. Pausing as precautionary measure.", "confidence": 70, } result_json = json.dumps(result) if result.get("recommend_pause"): self.paused_protocols[protocol] = True log_entry = json.dumps({ "action": "ai_pause", "protocol": protocol, "reason": result.get("reasoning", ""), "timestamp": str(gl.block.timestamp), "admin": "ai_evaluation", }) self.emergency_log.append(log_entry) warning = json.dumps({ "protocol": protocol, "severity": result.get("severity", "high"), "summary": result.get("reasoning", ""), "timestamp": str(gl.block.timestamp), }) self.warnings.append(warning) return result_json @gl.public.write def raise_warning(self, protocol: str, message: str, severity: str) -> None: assert gl.message.sender_address == self.admin, "Only admin" warning = json.dumps({ "protocol": protocol, "severity": severity, "summary": message, "timestamp": str(gl.block.timestamp), }) self.warnings.append(warning)