""" AIVA Constitutional Decision Engine ==================================== Constitutional decision engine for AIVA Queen. Given an action request, determines: - Which Tier it falls into (0, 1, 2, or 3) - Whether AIVA can proceed autonomously - What notification or approval is required - Logs decision to Knowledge Graph entity file CRITICAL: Does NOT connect to AIVA's server (152.53.201.152). Runs locally on E:\\genesis-system. Can be imported by AIVA when she has connectivity to Genesis. Governing document: E:\\genesis-system\\AIVA\\AIVA_CONSTITUTION.md Author: AIVA Constitutional Architect Version: 1.0.0 Ratified: 2026-02-23 """ import json import logging from datetime import datetime from pathlib import Path from typing import Optional from dataclasses import dataclass, asdict from enum import IntEnum # --------------------------------------------------------------------------- # Constants and paths # --------------------------------------------------------------------------- GENESIS_ROOT = Path("E:/genesis-system") KG_ENTITIES_DIR = GENESIS_ROOT / "KNOWLEDGE_GRAPH" / "entities" DECISION_LOG_FILE = KG_ENTITIES_DIR / "aiva_decision_engine_log.jsonl" # The Elestio server — UNTOUCHABLE. This engine never touches it. FORBIDDEN_HOST = "152.53.201.152" log = logging.getLogger("aiva.decision_engine") # --------------------------------------------------------------------------- # Tier definitions # --------------------------------------------------------------------------- class Tier(IntEnum): """ Constitutional authority tiers from AIVA_CONSTITUTION.md Article 2. TIER_0 — AIVA Autonomous: No approval needed. TIER_1 — Notify Kinan: Execute, then inform within 1 hour. TIER_2 — Approval Required: Must get explicit approval first. TIER_3 — Absolute Prohibition: Never, under any circumstances. """ TIER_0 = 0 # Autonomous TIER_1 = 1 # Notify after TIER_2 = 2 # Approval before TIER_3 = 3 # Permanently forbidden @dataclass class DecisionResult: """Result returned by classify_action().""" action: str tier: int tier_name: str can_proceed: bool requires_approval: bool notification_required: bool approval_timeout_seconds: Optional[int] constitutional_basis: str reasoning: str timestamp: str def to_dict(self) -> dict: return asdict(self) # --------------------------------------------------------------------------- # Tier 0 keyword catalogue # --------------------------------------------------------------------------- TIER_0_PATTERNS = { # Cognitive / response "cognitive": [ "respond", "answer", "reply", "generate analysis", "generate report", "create recommendation", "draft", "compose text", "write content for review", "form opinion", "evaluate options", "present tradeoff", ], # Memory reads "memory_read": [ "read memory", "query memory", "retrieve", "recall", "lookup", "search knowledge graph", "query postgresql", "query qdrant", "query redis", "fetch from memory", "read from database", ], # Memory writes (own working memory only) "memory_write_working": [ "update working memory", "write to redis", "set redis key", "update confidence score", "record task outcome", "log to working memory", "update familiarity", "write axiom", "store axiom", ], # Monitoring and reporting "monitoring": [ "health check", "monitor", "check service", "check infrastructure", "measure latency", "collect metrics", "log performance", "generate dashboard", "status summary", "report uptime", "alert kinan", "telegram alert", "send telegram notification", ], # Research "research": [ "background research", "web search", "competitor monitoring", "market intelligence", "extract axiom", "scan industry news", "public data query", "search internet", "research topic", ], # Internal orchestration "orchestration": [ "spawn gemini", "spawn sub-agent", "route task to", "dispatch task", "aggregate results", "coordinate agents", "execute openClaw skill", "run approved skill", "start cron job", ], # Mac Mini operations (local only) "mac_mini_local": [ "restart aiva process", "restart on mac mini", "deploy openClaw skill", "manage local model", "ollama localhost", "manage local ollama", "run scheduled task", "mac mini maintenance", ], } # --------------------------------------------------------------------------- # Tier 1 keyword catalogue # --------------------------------------------------------------------------- TIER_1_PATTERNS = { # Self-modification "self_modification": [ "update system prompt", "modify persona", "change conversation style", "update own configuration", "modify confidence weights", "add to skill registry", "create new openClaw skill", "deploy new skill", "modify skill registry", ], # Internal content "internal_content": [ "post to slack", "post internal report", "create file in genesis", "write to notion", "create genesis directory", "modify configuration file", "deploy workflow template", "create github branch", ], # Memory promotion "memory_promotion": [ "promote to episodic memory", "write to postgresql episodic", "consolidate memory", "archive old records", "prune redis entries", "memory cleanup", "episodic memory write", ], # Mac Mini infrastructure (not Elestio) "mac_mini_infra": [ "scale mac mini", "install package on mac mini", "modify openClaw config", "rotate log files", "mac mini configuration", "update mac mini settings", ], } # --------------------------------------------------------------------------- # Tier 2 keyword catalogue # --------------------------------------------------------------------------- TIER_2_PATTERNS = { # External communications "external_comms": [ "send email", "send sms", "post to social media", "post to twitter", "post to linkedin", "post to facebook", "publish publicly", "send message via ghl", "send via instantly", "outreach", "cold email", "send outbound message", "message lead", "contact prospect", ], # Financial "financial": [ "spend", "purchase", "buy", "pay", "transaction", "payment", "charge", "stripe", "subscription create", "subscription modify", "api credit", "top up balance", "approve budget", "allocate budget", "financial commitment", "invoice", "refund", "revenue reallocate", ], # New integrations "new_integration": [ "connect new api", "create api key", "new webhook", "establish integration", "add external service", "connect to service", "new n8n connection", "integrate with", "add new connector", ], # Data sharing "data_sharing": [ "export data", "share data externally", "send data outside genesis", "export customer data", "share ghl data", "transmit user data", "share with third party", ], # Significant system changes "system_changes": [ "deploy to production", "modify core architecture", "create database", "create new schema", "alter permissions", "change access control", "modify security settings", "production deployment", ], } # --------------------------------------------------------------------------- # Tier 3 absolute prohibitions # --------------------------------------------------------------------------- TIER_3_PATTERNS = { # SSH to Elestio — permanently banned "elestio_access": [ "ssh 152.53.201.152", "ssh to elestio", "connect elestio server", "access elestio", "modify elestio", "elestio ssh", "152.53.201.152", "pull ollama model on elestio", "delete ollama model on elestio", "restart elestio", "elestio server command", ], # Identity deception "identity_deception": [ "claim to be human", "say i am human", "deny being ai", "pretend to be human", "impersonate human", "deceive as human", ], # Irreversible data deletion "irreversible_deletion": [ "permanently delete production data", "drop table", "drop database", "delete all records", "wipe knowledge graph", "clear episodic memory", "destroy semantic memory", "delete production database", ], # Constitutional override "constitutional_override": [ "override constitution", "bypass hitl", "disable hitl gate", "ignore constitutional", "override tier", "modify constitution", "disable safety", "bypass safety rule", "skip approval", "proceed without approval", ], # Financial transactions without approval "unauthorized_transaction": [ "execute transaction", "make payment without approval", "charge card without approval", "initiate purchase without approval", ], # Memory fabrication "memory_fabrication": [ "fabricate memory", "create false memory", "alter memory record", "edit past interaction", "falsify log", "modify audit trail", ], } # --------------------------------------------------------------------------- # Decision timeout configuration (Tier 2) # --------------------------------------------------------------------------- TIER_2_TIMEOUTS = { "financial": 600, # 10 minutes "external_comms": 300, # 5 minutes "new_integration": None, # Wait indefinitely "data_sharing": 600, # 10 minutes "system_changes": 600, # 10 minutes "default": 600, # 10 minutes } # --------------------------------------------------------------------------- # Decision Engine # --------------------------------------------------------------------------- class AIVADecisionEngine: """ Constitutional decision engine for AIVA. Usage: engine = AIVADecisionEngine() result = engine.classify_action( action="send email to lead@example.com", context={"recipient": "lead@example.com", "purpose": "sales follow-up"} ) if result.can_proceed: # Execute action pass elif result.requires_approval: # Request Kinan approval via HITL gate pass else: # Tier 3 — do not proceed, log violation pass """ def __init__(self, log_decisions: bool = True): self.log_decisions = log_decisions self._ensure_log_file() def _ensure_log_file(self) -> None: """Ensure the KG entities directory and log file exist.""" try: KG_ENTITIES_DIR.mkdir(parents=True, exist_ok=True) except Exception as exc: log.warning("Could not create KG entities dir: %s", exc) # ------------------------------------------------------------------ # Public API # ------------------------------------------------------------------ def classify_action(self, action: str, context: Optional[dict] = None) -> DecisionResult: """ Classify an action into a constitutional Tier and determine whether AIVA may proceed. Args: action: Human-readable description of the action to classify. context: Optional dict with additional context (e.g., recipient, amount, target system). Returns: DecisionResult with tier, can_proceed, requires_approval, etc. """ action_lower = action.lower() context = context or {} # Safety check: never proceed with anything targeting 152.53.201.152 if self._check_elestio_reference(action_lower, context): result = self._make_result( action=action, tier=Tier.TIER_3, constitutional_basis="Article 1 Section 1.5 + Article 2 Section 2.4", reasoning=( "Action references Elestio server (152.53.201.152) which is permanently " "untouchable. No SSH, modification, or command execution is ever permitted " "on this host under any circumstances." ), ) self.log_decision(action, Tier.TIER_3, "BLOCKED — Elestio access attempt") return result # Check Tier 3 first (prohibitions take precedence) tier3_match = self._match_patterns(action_lower, TIER_3_PATTERNS) if tier3_match: category, matched_keyword = tier3_match result = self._make_result( action=action, tier=Tier.TIER_3, constitutional_basis="Article 2 Section 2.4", reasoning=( f"Action matches Tier 3 absolute prohibition category '{category}' " f"(matched keyword: '{matched_keyword}'). This action is permanently " "forbidden and cannot be executed under any circumstances." ), ) self.log_decision(action, Tier.TIER_3, f"BLOCKED — {category}") return result # Check Tier 2 (approval required) tier2_match = self._match_patterns(action_lower, TIER_2_PATTERNS) if tier2_match: category, matched_keyword = tier2_match timeout = TIER_2_TIMEOUTS.get(category, TIER_2_TIMEOUTS["default"]) result = self._make_result( action=action, tier=Tier.TIER_2, constitutional_basis="Article 2 Section 2.3", reasoning=( f"Action matches Tier 2 category '{category}' " f"(matched keyword: '{matched_keyword}'). " "Kinan approval is required before proceeding. " f"Timeout: {timeout} seconds (auto-reject if no response)." ), tier2_category=category, approval_timeout=timeout, ) self.log_decision(action, Tier.TIER_2, f"PENDING APPROVAL — {category}") return result # Check Tier 1 (notify after) tier1_match = self._match_patterns(action_lower, TIER_1_PATTERNS) if tier1_match: category, matched_keyword = tier1_match result = self._make_result( action=action, tier=Tier.TIER_1, constitutional_basis="Article 2 Section 2.2", reasoning=( f"Action matches Tier 1 category '{category}' " f"(matched keyword: '{matched_keyword}'). " "AIVA may proceed now and must notify Kinan within 1 hour." ), ) self.log_decision(action, Tier.TIER_1, f"PROCEED + NOTIFY — {category}") return result # Check Tier 0 (fully autonomous) tier0_match = self._match_patterns(action_lower, TIER_0_PATTERNS) if tier0_match: category, matched_keyword = tier0_match result = self._make_result( action=action, tier=Tier.TIER_0, constitutional_basis="Article 2 Section 2.1", reasoning=( f"Action matches Tier 0 category '{category}' " f"(matched keyword: '{matched_keyword}'). " "AIVA may proceed autonomously. No notification required." ), ) self.log_decision(action, Tier.TIER_0, f"AUTONOMOUS — {category}") return result # No pattern matched — default to Tier 2 (safe escalation) result = self._make_result( action=action, tier=Tier.TIER_2, constitutional_basis="Article 3 Section 3.1 (uncertainty protocol)", reasoning=( "Action did not match any known Tier pattern. Per Article 3 Section 3.1, " "when uncertain, AIVA escalates to the most restrictive plausible Tier. " "Defaulting to Tier 2 — Kinan approval required. AIVA must not proceed until " "the action is explicitly classified and approved." ), tier2_category="unclassified", approval_timeout=TIER_2_TIMEOUTS["default"], ) self.log_decision(action, Tier.TIER_2, "PENDING APPROVAL — unclassified (safe default)") return result def check_prohibitions(self, action: str) -> bool: """ Fast check: returns True if the action is absolutely prohibited (Tier 3). Use this for quick guard checks before any action execution. Args: action: Human-readable description of the action. Returns: True if action is prohibited, False if it may be permitted. """ action_lower = action.lower() if self._check_elestio_reference(action_lower, {}): return True tier3_match = self._match_patterns(action_lower, TIER_3_PATTERNS) return tier3_match is not None def get_tier_name(self, tier: int) -> str: """Return the human-readable name for a Tier integer.""" names = { 0: "Tier 0 — AIVA Autonomous", 1: "Tier 1 — Notify Kinan", 2: "Tier 2 — Kinan Approval Required", 3: "Tier 3 — Absolute Prohibition", } return names.get(tier, f"Unknown Tier {tier}") def log_decision(self, action: str, tier: int, decision: str) -> None: """ Log a decision to the KG entity file. Format: one JSON object per line (JSONL). File: KNOWLEDGE_GRAPH/entities/aiva_decision_engine_log.jsonl Args: action: The action that was classified. tier: The Tier assigned (0-3). decision: Short description of the outcome. """ if not self.log_decisions: return entry = { "id": f"decision_{datetime.utcnow().strftime('%Y%m%d_%H%M%S_%f')}", "timestamp": datetime.utcnow().isoformat() + "Z", "action": action, "tier": tier, "tier_name": self.get_tier_name(tier), "decision": decision, "source": "aiva_decision_engine.py", } try: with open(DECISION_LOG_FILE, "a", encoding="utf-8") as fh: fh.write(json.dumps(entry) + "\n") except Exception as exc: log.error("Failed to log decision to KG entities: %s", exc) # ------------------------------------------------------------------ # Private helpers # ------------------------------------------------------------------ def _check_elestio_reference(self, action_lower: str, context: dict) -> bool: """ Check whether the action or context references the Elestio server. The Elestio host (152.53.201.152) is permanently untouchable. Any reference to it in an action description triggers Tier 3. """ elestio_markers = [ "152.53.201.152", "elestio server", "elestio host", "ssh elestio", ] for marker in elestio_markers: if marker in action_lower: return True # Check context values for value in context.values(): if isinstance(value, str) and "152.53.201.152" in value: return True return False def _match_patterns( self, action_lower: str, pattern_dict: dict, ) -> Optional[tuple[str, str]]: """ Search pattern_dict for any keyword appearing in action_lower. Returns: Tuple of (category_name, matched_keyword) if found, else None. """ for category, keywords in pattern_dict.items(): for keyword in keywords: if keyword in action_lower: return category, keyword return None def _make_result( self, action: str, tier: int, constitutional_basis: str, reasoning: str, tier2_category: Optional[str] = None, approval_timeout: Optional[int] = None, ) -> DecisionResult: """Construct a DecisionResult from classification data.""" return DecisionResult( action=action, tier=tier, tier_name=self.get_tier_name(tier), can_proceed=(tier in (Tier.TIER_0, Tier.TIER_1)), requires_approval=(tier == Tier.TIER_2), notification_required=(tier == Tier.TIER_1), approval_timeout_seconds=approval_timeout, constitutional_basis=constitutional_basis, reasoning=reasoning, timestamp=datetime.utcnow().isoformat() + "Z", ) # --------------------------------------------------------------------------- # Convenience factory # --------------------------------------------------------------------------- def get_decision_engine(log_decisions: bool = True) -> AIVADecisionEngine: """Return a shared decision engine instance.""" return AIVADecisionEngine(log_decisions=log_decisions) # --------------------------------------------------------------------------- # CLI for quick classification checks # --------------------------------------------------------------------------- if __name__ == "__main__": import sys if len(sys.argv) < 2: print("Usage: python aiva_decision_engine.py \"action description\"") print("Example: python aiva_decision_engine.py \"send email to lead@example.com\"") sys.exit(1) action_input = " ".join(sys.argv[1:]) engine = AIVADecisionEngine(log_decisions=True) result = engine.classify_action(action_input) print("\n" + "=" * 60) print("AIVA CONSTITUTIONAL DECISION ENGINE") print("=" * 60) print(f"Action: {result.action}") print(f"Tier: {result.tier} — {result.tier_name}") print(f"Can Proceed: {result.can_proceed}") print(f"Requires Approval: {result.requires_approval}") print(f"Notification Required:{result.notification_required}") if result.approval_timeout_seconds: print(f"Approval Timeout: {result.approval_timeout_seconds}s") print(f"Constitutional Basis: {result.constitutional_basis}") print(f"Reasoning:") print(f" {result.reasoning}") print(f"Timestamp: {result.timestamp}") print("=" * 60 + "\n")