""" OpenWork Action Approval Workflow ================================= Safety layer for high-risk OpenWork actions with Kinan approval integration. This module provides: - Risk classification for OpenWork actions - Auto-approval for low-risk actions - Kinan approval flow for high-risk actions - Action audit logging - Configurable approval policies Author: Genesis System Version: 1.0.0 """ import asyncio import json import logging import uuid from dataclasses import dataclass, field from datetime import datetime, timedelta from enum import Enum, auto from pathlib import Path from typing import Any, Callable, Dict, List, Optional, Set, Union logger = logging.getLogger(__name__) class RiskLevel(Enum): """Risk levels for OpenWork actions.""" MINIMAL = 0 # No approval needed, instant execution LOW = 1 # Auto-approve with logging MEDIUM = 2 # Auto-approve with delay + notification HIGH = 3 # Requires Kinan approval CRITICAL = 4 # Requires explicit Kinan approval, no auto-approve class ActionCategory(Enum): """Categories of OpenWork actions.""" FILE_READ = auto() FILE_WRITE = auto() FILE_DELETE = auto() BROWSER_NAVIGATE = auto() BROWSER_CLICK = auto() BROWSER_INPUT = auto() BROWSER_CREDENTIALS = auto() CLIPBOARD = auto() NOTIFICATION = auto() SYSTEM_COMMAND = auto() APPLICATION_LAUNCH = auto() APPLICATION_CLOSE = auto() NETWORK_REQUEST = auto() KEYBOARD_INPUT = auto() SCREENSHOT = auto() UNKNOWN = auto() class ApprovalDecision(Enum): """Possible approval decisions.""" PENDING = auto() APPROVED = auto() REJECTED = auto() EXPIRED = auto() AUTO_APPROVED = auto() @dataclass class ActionRiskProfile: """Risk profile for an action type.""" category: ActionCategory base_risk: RiskLevel modifiers: Dict[str, RiskLevel] = field(default_factory=dict) requires_approval_above: RiskLevel = RiskLevel.MEDIUM max_auto_approve_count: int = 10 # Per hour cooldown_seconds: int = 0 @dataclass class ApprovalRequest: """Request for action approval.""" request_id: str action_type: str action_category: ActionCategory risk_level: RiskLevel payload: Dict[str, Any] justification: str context: Dict[str, Any] decision: ApprovalDecision = ApprovalDecision.PENDING created_at: datetime = field(default_factory=datetime.utcnow) decided_at: Optional[datetime] = None decision_reason: Optional[str] = None approver: Optional[str] = None expires_at: Optional[datetime] = None def to_dict(self) -> Dict[str, Any]: """Serialize to dictionary.""" return { "request_id": self.request_id, "action_type": self.action_type, "action_category": self.action_category.name, "risk_level": self.risk_level.name, "payload": self.payload, "justification": self.justification, "context": self.context, "decision": self.decision.name, "created_at": self.created_at.isoformat(), "decided_at": self.decided_at.isoformat() if self.decided_at else None, "decision_reason": self.decision_reason, "approver": self.approver, "expires_at": self.expires_at.isoformat() if self.expires_at else None } @property def is_expired(self) -> bool: """Check if request has expired.""" if not self.expires_at: return False return datetime.utcnow() > self.expires_at and self.decision == ApprovalDecision.PENDING @dataclass class ActionAuditEntry: """Audit log entry for an action.""" entry_id: str request_id: str action_type: str risk_level: str decision: str timestamp: datetime details: Dict[str, Any] def to_dict(self) -> Dict[str, Any]: """Serialize to dictionary.""" return { "entry_id": self.entry_id, "request_id": self.request_id, "action_type": self.action_type, "risk_level": self.risk_level, "decision": self.decision, "timestamp": self.timestamp.isoformat(), "details": self.details } class RiskClassifier: """ Classifies actions by risk level based on type and context. """ # Default risk profiles for action categories DEFAULT_PROFILES: Dict[ActionCategory, ActionRiskProfile] = { ActionCategory.FILE_READ: ActionRiskProfile( category=ActionCategory.FILE_READ, base_risk=RiskLevel.LOW, modifiers={ "sensitive_path": RiskLevel.HIGH, "credentials": RiskLevel.CRITICAL } ), ActionCategory.FILE_WRITE: ActionRiskProfile( category=ActionCategory.FILE_WRITE, base_risk=RiskLevel.MEDIUM, modifiers={ "system_path": RiskLevel.HIGH, "executable": RiskLevel.CRITICAL } ), ActionCategory.FILE_DELETE: ActionRiskProfile( category=ActionCategory.FILE_DELETE, base_risk=RiskLevel.HIGH, modifiers={ "system_path": RiskLevel.CRITICAL } ), ActionCategory.BROWSER_NAVIGATE: ActionRiskProfile( category=ActionCategory.BROWSER_NAVIGATE, base_risk=RiskLevel.LOW ), ActionCategory.BROWSER_CLICK: ActionRiskProfile( category=ActionCategory.BROWSER_CLICK, base_risk=RiskLevel.LOW ), ActionCategory.BROWSER_INPUT: ActionRiskProfile( category=ActionCategory.BROWSER_INPUT, base_risk=RiskLevel.MEDIUM, modifiers={ "password_field": RiskLevel.HIGH } ), ActionCategory.BROWSER_CREDENTIALS: ActionRiskProfile( category=ActionCategory.BROWSER_CREDENTIALS, base_risk=RiskLevel.CRITICAL ), ActionCategory.CLIPBOARD: ActionRiskProfile( category=ActionCategory.CLIPBOARD, base_risk=RiskLevel.MINIMAL ), ActionCategory.NOTIFICATION: ActionRiskProfile( category=ActionCategory.NOTIFICATION, base_risk=RiskLevel.MINIMAL ), ActionCategory.SYSTEM_COMMAND: ActionRiskProfile( category=ActionCategory.SYSTEM_COMMAND, base_risk=RiskLevel.CRITICAL ), ActionCategory.APPLICATION_LAUNCH: ActionRiskProfile( category=ActionCategory.APPLICATION_LAUNCH, base_risk=RiskLevel.MEDIUM ), ActionCategory.APPLICATION_CLOSE: ActionRiskProfile( category=ActionCategory.APPLICATION_CLOSE, base_risk=RiskLevel.MEDIUM ), ActionCategory.NETWORK_REQUEST: ActionRiskProfile( category=ActionCategory.NETWORK_REQUEST, base_risk=RiskLevel.MEDIUM, modifiers={ "external_api": RiskLevel.HIGH } ), ActionCategory.KEYBOARD_INPUT: ActionRiskProfile( category=ActionCategory.KEYBOARD_INPUT, base_risk=RiskLevel.LOW ), ActionCategory.SCREENSHOT: ActionRiskProfile( category=ActionCategory.SCREENSHOT, base_risk=RiskLevel.LOW ), ActionCategory.UNKNOWN: ActionRiskProfile( category=ActionCategory.UNKNOWN, base_risk=RiskLevel.HIGH ) } # Sensitive path patterns SENSITIVE_PATHS = { "credentials": [".env", "credentials", "secret", ".pem", ".key", "password"], "system": ["/etc/", "/usr/", "/bin/", "C:\\Windows\\", "C:\\Program Files\\"], "config": [".ssh", ".aws", ".azure", ".gcloud"] } def __init__(self, custom_profiles: Optional[Dict[ActionCategory, ActionRiskProfile]] = None): self.profiles = {**self.DEFAULT_PROFILES} if custom_profiles: self.profiles.update(custom_profiles) def classify( self, action_type: str, payload: Dict[str, Any], context: Optional[Dict[str, Any]] = None ) -> tuple[ActionCategory, RiskLevel]: """ Classify an action and determine its risk level. Args: action_type: The type of action payload: Action payload/parameters context: Additional context Returns: Tuple of (ActionCategory, RiskLevel) """ category = self._determine_category(action_type, payload) base_risk = self.profiles[category].base_risk # Apply modifiers based on payload content final_risk = self._apply_modifiers(category, base_risk, payload, context or {}) return category, final_risk def _determine_category(self, action_type: str, payload: Dict[str, Any]) -> ActionCategory: """Determine action category from type.""" action_lower = action_type.lower() # File operations if any(kw in action_lower for kw in ["file_read", "read_file", "open_file"]): return ActionCategory.FILE_READ if any(kw in action_lower for kw in ["file_write", "write_file", "save_file", "create_file"]): return ActionCategory.FILE_WRITE if any(kw in action_lower for kw in ["file_delete", "delete_file", "remove_file"]): return ActionCategory.FILE_DELETE # Browser operations if any(kw in action_lower for kw in ["navigate", "goto", "open_url"]): return ActionCategory.BROWSER_NAVIGATE if any(kw in action_lower for kw in ["click", "press"]): return ActionCategory.BROWSER_CLICK if any(kw in action_lower for kw in ["input", "type", "fill"]): if self._is_credential_field(payload): return ActionCategory.BROWSER_CREDENTIALS return ActionCategory.BROWSER_INPUT # System operations if any(kw in action_lower for kw in ["command", "shell", "exec", "run"]): return ActionCategory.SYSTEM_COMMAND if any(kw in action_lower for kw in ["launch", "start_app", "open_app"]): return ActionCategory.APPLICATION_LAUNCH if any(kw in action_lower for kw in ["close_app", "quit", "kill"]): return ActionCategory.APPLICATION_CLOSE # Other operations if "clipboard" in action_lower: return ActionCategory.CLIPBOARD if any(kw in action_lower for kw in ["notify", "notification", "alert"]): return ActionCategory.NOTIFICATION if any(kw in action_lower for kw in ["request", "fetch", "api"]): return ActionCategory.NETWORK_REQUEST if any(kw in action_lower for kw in ["keyboard", "key_press"]): return ActionCategory.KEYBOARD_INPUT if "screenshot" in action_lower: return ActionCategory.SCREENSHOT return ActionCategory.UNKNOWN def _is_credential_field(self, payload: Dict[str, Any]) -> bool: """Check if payload targets a credential field.""" field_name = str(payload.get("field", "")).lower() element_type = str(payload.get("type", "")).lower() credential_indicators = ["password", "secret", "token", "api_key", "credential"] return ( element_type == "password" or any(ind in field_name for ind in credential_indicators) ) def _apply_modifiers( self, category: ActionCategory, base_risk: RiskLevel, payload: Dict[str, Any], context: Dict[str, Any] ) -> RiskLevel: """Apply risk modifiers based on payload and context.""" final_risk = base_risk profile = self.profiles[category] # Check for sensitive paths in file operations if category in [ActionCategory.FILE_READ, ActionCategory.FILE_WRITE, ActionCategory.FILE_DELETE]: path = str(payload.get("path", payload.get("file_path", ""))).lower() for pattern_type, patterns in self.SENSITIVE_PATHS.items(): if any(p in path for p in patterns): modifier_risk = profile.modifiers.get( pattern_type, profile.modifiers.get("sensitive_path", RiskLevel.HIGH) ) if modifier_risk.value > final_risk.value: final_risk = modifier_risk # Check for credentials context if context.get("involves_credentials"): if profile.modifiers.get("credentials"): final_risk = profile.modifiers["credentials"] # Check for external network requests if category == ActionCategory.NETWORK_REQUEST: url = str(payload.get("url", "")).lower() if not any(local in url for local in ["localhost", "127.0.0.1", "genesis"]): modifier_risk = profile.modifiers.get("external_api", RiskLevel.HIGH) if modifier_risk.value > final_risk.value: final_risk = modifier_risk return final_risk class ApprovalWorkflow: """ Manages the approval workflow for OpenWork actions. Integrates with KinanLiaisonSystem for high-risk action approvals. """ def __init__( self, config: Optional[Dict[str, Any]] = None, kinan_liaison: Optional[Any] = None # KinanLiaisonSystem ): self.config = config or {} self.kinan_liaison = kinan_liaison self.classifier = RiskClassifier() # Pending approval requests self._pending: Dict[str, ApprovalRequest] = {} # Audit log self._audit_log: List[ActionAuditEntry] = [] # Rate limiting self._action_counts: Dict[str, List[datetime]] = {} # Callbacks for approval decisions self._approval_callbacks: Dict[str, Callable] = {} # Auto-approve settings from config self._auto_approve_low_risk = config.get("auto_approve_low_risk", True) self._auto_approve_timeout_hours = config.get("auto_approve_timeout_hours", 48) self._low_risk_actions = set(config.get("low_risk_actions", [ "notification", "clipboard", "file_read" ])) self._high_risk_actions = set(config.get("high_risk_actions", [ "file_delete", "system_command", "browser_credentials" ])) # Background task for expiration checking self._expiration_task: Optional[asyncio.Task] = None self._is_running = False logger.info("ApprovalWorkflow initialized") async def start(self): """Start the approval workflow background tasks.""" if self._is_running: return self._is_running = True self._expiration_task = asyncio.create_task(self._check_expirations()) logger.info("ApprovalWorkflow started") async def stop(self): """Stop the approval workflow.""" self._is_running = False if self._expiration_task: self._expiration_task.cancel() try: await self._expiration_task except asyncio.CancelledError: pass logger.info("ApprovalWorkflow stopped") async def _check_expirations(self): """Background task to check for expired requests.""" while self._is_running: try: now = datetime.utcnow() expired_ids = [] for request_id, request in self._pending.items(): if request.is_expired: expired_ids.append(request_id) elif request.expires_at and now > request.expires_at: # Auto-approve if configured if self._should_auto_approve_on_expiry(request): await self._auto_approve(request_id, "Auto-approved on timeout") else: expired_ids.append(request_id) for request_id in expired_ids: await self._expire_request(request_id) await asyncio.sleep(60) # Check every minute except asyncio.CancelledError: break except Exception as e: logger.error(f"Expiration check error: {e}") await asyncio.sleep(60) def _should_auto_approve_on_expiry(self, request: ApprovalRequest) -> bool: """Determine if request should auto-approve on timeout.""" # Never auto-approve critical actions if request.risk_level == RiskLevel.CRITICAL: return False # Check if action type is in low-risk list if request.action_type.lower() in self._low_risk_actions: return True # Check if risk level allows auto-approval return request.risk_level.value <= RiskLevel.MEDIUM.value async def request_approval( self, action_type: str, payload: Dict[str, Any], justification: str = "", context: Optional[Dict[str, Any]] = None, callback: Optional[Callable] = None ) -> ApprovalRequest: """ Request approval for an action. Args: action_type: Type of action payload: Action payload justification: Why this action is needed context: Additional context callback: Function to call when decision is made Returns: ApprovalRequest object """ context = context or {} # Classify the action category, risk_level = self.classifier.classify(action_type, payload, context) # Create request request_id = str(uuid.uuid4()) expires_at = datetime.utcnow() + timedelta(hours=self._auto_approve_timeout_hours) request = ApprovalRequest( request_id=request_id, action_type=action_type, action_category=category, risk_level=risk_level, payload=payload, justification=justification, context=context, expires_at=expires_at ) # Store callback if provided if callback: self._approval_callbacks[request_id] = callback # Check if auto-approval is possible if await self._can_auto_approve(request): await self._auto_approve(request_id, "Low-risk action auto-approved", request) return request # Store pending request self._pending[request_id] = request # Request Kinan approval for high-risk actions if self.kinan_liaison and risk_level.value >= RiskLevel.HIGH.value: await self._request_kinan_approval(request) # Log the request self._log_action(request, "PENDING") logger.info(f"Approval requested: {request_id} - {action_type} (Risk: {risk_level.name})") return request async def _can_auto_approve(self, request: ApprovalRequest) -> bool: """Check if action can be auto-approved.""" if not self._auto_approve_low_risk: return False # Never auto-approve critical or high-risk if request.risk_level.value >= RiskLevel.HIGH.value: return False # Check rate limits if not self._check_rate_limit(request.action_category): return False # Check if action type is explicitly high-risk if request.action_type.lower() in self._high_risk_actions: return False return True def _check_rate_limit(self, category: ActionCategory) -> bool: """Check if action is within rate limits.""" key = category.name now = datetime.utcnow() hour_ago = now - timedelta(hours=1) # Clean old entries if key in self._action_counts: self._action_counts[key] = [ ts for ts in self._action_counts[key] if ts > hour_ago ] else: self._action_counts[key] = [] profile = self.classifier.profiles.get(category) max_count = profile.max_auto_approve_count if profile else 10 return len(self._action_counts[key]) < max_count async def _auto_approve( self, request_id: str, reason: str, request: Optional[ApprovalRequest] = None ): """Auto-approve a request.""" if request is None: request = self._pending.get(request_id) if request: request.decision = ApprovalDecision.AUTO_APPROVED request.decided_at = datetime.utcnow() request.decision_reason = reason request.approver = "SYSTEM" # Track for rate limiting key = request.action_category.name if key not in self._action_counts: self._action_counts[key] = [] self._action_counts[key].append(datetime.utcnow()) # Remove from pending if request_id in self._pending: del self._pending[request_id] # Trigger callback if request_id in self._approval_callbacks: callback = self._approval_callbacks.pop(request_id) try: if asyncio.iscoroutinefunction(callback): await callback(request) else: callback(request) except Exception as e: logger.error(f"Callback error: {e}") # Log self._log_action(request, "AUTO_APPROVED") logger.info(f"Auto-approved: {request_id}") async def _request_kinan_approval(self, request: ApprovalRequest): """Request approval from Kinan via the liaison system.""" if not self.kinan_liaison: logger.warning("Kinan liaison not configured - action pending manual approval") return try: # Format payload for readability payload_summary = self._format_payload_summary(request.payload) # Build impact assessment impact = { "action_type": request.action_type, "category": request.action_category.name, "risk_level": request.risk_level.name, "target": request.payload.get("path", request.payload.get("url", "N/A")) } # Build risk analysis risk = { "risk_level": request.risk_level.name, "reversible": self._is_reversible(request), "scope": self._determine_scope(request) } # Request approval through liaison await self.kinan_liaison.approvals.seek_approval( action_description=f"OpenWork Action: {request.action_type}\n\n{payload_summary}", impact_assessment=impact, risk_analysis=risk, recommended_action=request.justification or "Proceed if intended", deadline=request.expires_at, auto_approve_hours=self._auto_approve_timeout_hours if request.risk_level != RiskLevel.CRITICAL else None, category="openwork_action" ) logger.info(f"Kinan approval requested for: {request.request_id}") except Exception as e: logger.error(f"Failed to request Kinan approval: {e}") def _format_payload_summary(self, payload: Dict[str, Any]) -> str: """Format payload for human readability.""" lines = [] for key, value in payload.items(): if isinstance(value, dict): value = json.dumps(value, indent=2) lines.append(f" {key}: {value}") return "\n".join(lines) if lines else " (empty payload)" def _is_reversible(self, request: ApprovalRequest) -> str: """Determine if action is reversible.""" reversible_categories = { ActionCategory.FILE_READ, ActionCategory.BROWSER_NAVIGATE, ActionCategory.CLIPBOARD, ActionCategory.NOTIFICATION, ActionCategory.SCREENSHOT } if request.action_category in reversible_categories: return "Yes" elif request.action_category == ActionCategory.FILE_DELETE: return "No (unless backed up)" elif request.action_category == ActionCategory.FILE_WRITE: return "Partial (overwritten data lost)" else: return "Depends on action" def _determine_scope(self, request: ApprovalRequest) -> str: """Determine the scope of the action.""" if request.action_category in [ActionCategory.SYSTEM_COMMAND]: return "System-wide" elif request.action_category in [ActionCategory.FILE_DELETE, ActionCategory.FILE_WRITE]: return "File system" elif request.action_category.name.startswith("BROWSER"): return "Browser session" else: return "Limited" async def approve( self, request_id: str, approver: str = "Kinan", reason: str = "" ) -> bool: """ Approve a pending request. Args: request_id: The request to approve approver: Who approved it reason: Optional approval reason Returns: True if approved, False if request not found """ if request_id not in self._pending: logger.warning(f"Approval request not found: {request_id}") return False request = self._pending[request_id] request.decision = ApprovalDecision.APPROVED request.decided_at = datetime.utcnow() request.decision_reason = reason request.approver = approver del self._pending[request_id] # Trigger callback if request_id in self._approval_callbacks: callback = self._approval_callbacks.pop(request_id) try: if asyncio.iscoroutinefunction(callback): await callback(request) else: callback(request) except Exception as e: logger.error(f"Callback error: {e}") # Log self._log_action(request, "APPROVED") logger.info(f"Approved by {approver}: {request_id}") return True async def reject( self, request_id: str, rejector: str = "Kinan", reason: str = "" ) -> bool: """ Reject a pending request. Args: request_id: The request to reject rejector: Who rejected it reason: Rejection reason Returns: True if rejected, False if request not found """ if request_id not in self._pending: logger.warning(f"Approval request not found: {request_id}") return False request = self._pending[request_id] request.decision = ApprovalDecision.REJECTED request.decided_at = datetime.utcnow() request.decision_reason = reason request.approver = rejector del self._pending[request_id] # Trigger callback with rejection if request_id in self._approval_callbacks: callback = self._approval_callbacks.pop(request_id) try: if asyncio.iscoroutinefunction(callback): await callback(request) else: callback(request) except Exception as e: logger.error(f"Callback error: {e}") # Log self._log_action(request, "REJECTED") logger.info(f"Rejected by {rejector}: {request_id} - {reason}") return True async def _expire_request(self, request_id: str): """Mark a request as expired.""" if request_id not in self._pending: return request = self._pending[request_id] request.decision = ApprovalDecision.EXPIRED request.decided_at = datetime.utcnow() request.decision_reason = "Request expired without decision" del self._pending[request_id] # Clean up callback self._approval_callbacks.pop(request_id, None) # Log self._log_action(request, "EXPIRED") logger.info(f"Request expired: {request_id}") def _log_action(self, request: ApprovalRequest, decision: str): """Log an action to the audit trail.""" entry = ActionAuditEntry( entry_id=str(uuid.uuid4()), request_id=request.request_id, action_type=request.action_type, risk_level=request.risk_level.name, decision=decision, timestamp=datetime.utcnow(), details={ "category": request.action_category.name, "payload_keys": list(request.payload.keys()), "justification": request.justification, "approver": request.approver } ) self._audit_log.append(entry) # Keep audit log size manageable (last 1000 entries) if len(self._audit_log) > 1000: self._audit_log = self._audit_log[-1000:] def get_pending_requests(self) -> List[ApprovalRequest]: """Get all pending approval requests.""" return list(self._pending.values()) def get_request(self, request_id: str) -> Optional[ApprovalRequest]: """Get a specific request by ID.""" return self._pending.get(request_id) def get_audit_log( self, limit: int = 100, action_type: Optional[str] = None, risk_level: Optional[str] = None ) -> List[ActionAuditEntry]: """ Get audit log entries with optional filtering. Args: limit: Maximum entries to return action_type: Filter by action type risk_level: Filter by risk level Returns: List of audit entries """ entries = self._audit_log.copy() if action_type: entries = [e for e in entries if e.action_type == action_type] if risk_level: entries = [e for e in entries if e.risk_level == risk_level] return entries[-limit:] def get_statistics(self) -> Dict[str, Any]: """Get approval workflow statistics.""" total = len(self._audit_log) by_decision = {} by_risk = {} by_category = {} for entry in self._audit_log: by_decision[entry.decision] = by_decision.get(entry.decision, 0) + 1 by_risk[entry.risk_level] = by_risk.get(entry.risk_level, 0) + 1 category = entry.details.get("category", "UNKNOWN") by_category[category] = by_category.get(category, 0) + 1 return { "total_actions": total, "pending_requests": len(self._pending), "by_decision": by_decision, "by_risk_level": by_risk, "by_category": by_category, "auto_approve_enabled": self._auto_approve_low_risk } # Convenience function to create workflow from config def create_approval_workflow( config_path: Optional[str] = None, kinan_liaison: Optional[Any] = None ) -> ApprovalWorkflow: """ Create an ApprovalWorkflow from configuration. Args: config_path: Path to openwork_config.json kinan_liaison: Optional KinanLiaisonSystem instance Returns: Configured ApprovalWorkflow """ config = {} if config_path: config_file = Path(config_path) if config_file.exists(): with open(config_file) as f: full_config = json.load(f) config = full_config.get("actions", {}).get("approval", {}) return ApprovalWorkflow(config=config, kinan_liaison=kinan_liaison) # Example usage async def example_usage(): """Demonstrate the approval workflow.""" # Create workflow workflow = ApprovalWorkflow( config={ "auto_approve_low_risk": True, "auto_approve_timeout_hours": 48, "low_risk_actions": ["notification", "clipboard", "file_read"], "high_risk_actions": ["file_delete", "system_command"] } ) await workflow.start() # Test low-risk action (should auto-approve) print("\n=== Testing Low-Risk Action ===") request1 = await workflow.request_approval( action_type="notification", payload={"message": "Task completed"}, justification="Notify user of completion" ) print(f"Request: {request1.request_id[:8]}") print(f"Decision: {request1.decision.name}") # Test high-risk action (should require approval) print("\n=== Testing High-Risk Action ===") request2 = await workflow.request_approval( action_type="file_delete", payload={"path": "/mnt/e/genesis-system/temp/old_file.txt"}, justification="Clean up temporary files" ) print(f"Request: {request2.request_id[:8]}") print(f"Decision: {request2.decision.name}") print(f"Risk Level: {request2.risk_level.name}") # Approve the high-risk action await workflow.approve(request2.request_id, "Kinan", "Approved cleanup") print(f"After approval: {request2.decision.name}") # Get statistics print("\n=== Statistics ===") stats = workflow.get_statistics() print(json.dumps(stats, indent=2)) await workflow.stop() if __name__ == "__main__": logging.basicConfig(level=logging.INFO) asyncio.run(example_usage())