#!/usr/bin/env python3 """ GENESIS EVOLUTION ENGINE ========================= Implements RULE 14: Failure Is Our Greatest Ally. Every failure that enters this engine is processed through the full 4-step Evolution Protocol: Step 1: DIAGNOSE — What happened, exactly? Step 2: ROOT CAUSE — Why? (5-Why analysis) Step 3: PRE-MORTEM — Guardrail to prevent recurrence Step 4: EVOLVE — Write to KG, axioms, CLAUDE.md, GLOBAL_GENESIS_RULES.md Also implements: - TITAN MEMORY updates (CLAUDE.md Titan Memory section) - RLM Worker pattern (watchful, associative, insatiable, fast) - Failure log persistence to KNOWLEDGE_GRAPH/entities/failure_evolution_log.jsonl - Axiom generation to KNOWLEDGE_GRAPH/axioms/ Usage: from core.evolution_engine import EvolutionEngine engine = EvolutionEngine() engine.process_failure({ "project": "receptionistai", "task": "RAI-003", "what": "Telnyx API returned 422 on assistant clone", "root_cause": "" }) # Also callable from CLI: python core/evolution_engine.py --failure "Gemini rate limit exceeded" """ import json import time import sys import re from datetime import datetime from pathlib import Path from typing import Dict, List, Optional, Any # All on E: drive (RULE 6) GENESIS_ROOT = Path("E:/genesis-system") KG_ENTITIES_PATH = GENESIS_ROOT / "KNOWLEDGE_GRAPH" / "entities" KG_AXIOMS_PATH = GENESIS_ROOT / "KNOWLEDGE_GRAPH" / "axioms" FAILURE_LOG_PATH = KG_ENTITIES_PATH / "failure_evolution_log.jsonl" CLAUDE_MD_PATH = GENESIS_ROOT / "CLAUDE.md" GLOBAL_RULES_PATH = GENESIS_ROOT / ".claude" / "rules" / "GLOBAL_GENESIS_RULES.md" # Surprise threshold — events scoring above this trigger full evolution SURPRISE_THRESHOLD = 0.50 # Domains where failures are most costly HIGH_STAKES_DOMAINS = [ "receptionistai", "aiva", "sunaiva", "revenue", "client", "telnyx", "stripe", "ghl", "voice", "production" ] class EvolutionEngine: """ Processes failures and surprises through the Genesis Evolution Protocol. This is the LEARNING brain of Genesis. Every failure makes Genesis smarter. The engine: 1. Diagnoses failures with structured analysis 2. Performs 5-Why root cause analysis 3. Designs guardrails to prevent recurrence 4. Writes findings to the living knowledge base 5. Updates TITAN MEMORY in CLAUDE.md 6. Generates axioms for the RLM bloodstream RULE 14 compliance is mandatory. This engine is NOT optional. """ def __init__(self): KG_ENTITIES_PATH.mkdir(parents=True, exist_ok=True) KG_AXIOMS_PATH.mkdir(parents=True, exist_ok=True) self.failures_processed = 0 self.axioms_generated = 0 def process_failure(self, failure: Dict[str, Any]) -> Dict[str, Any]: """ Main entry point. Process a failure through all 4 evolution steps. Args: failure: Dict with keys: project, task, what, root_cause (optional) Returns: Enriched failure dict with diagnosis, guardrail, axiom """ # Step 1: Diagnose failure = self._step1_diagnose(failure) # Step 2: Root cause (5-Why) failure = self._step2_root_cause(failure) # Step 3: Pre-mortem / guardrail failure = self._step3_premortem(failure) # Step 4: Evolve the system self._step4_evolve(failure) self.failures_processed += 1 return failure def _step1_diagnose(self, failure: Dict) -> Dict: """Step 1: Clarify exactly what happened.""" failure.setdefault("id", f"failure_{int(time.time())}") failure.setdefault("date", datetime.now().strftime("%Y-%m-%d")) failure.setdefault("timestamp", datetime.now().isoformat()) failure.setdefault("project", "unknown") failure.setdefault("task", "unknown") failure.setdefault("what", "Unspecified failure") # Score severity based on domain and keywords failure["severity"] = self._score_severity(failure) failure["surprise_score"] = self._score_surprise(failure) return failure def _step2_root_cause(self, failure: Dict) -> Dict: """Step 2: 5-Why root cause analysis (rule-based heuristics).""" what = failure.get("what", "").lower() # Pattern-match common root causes root_cause = failure.get("root_cause", "") if not root_cause or root_cause == "Not yet determined": root_cause = self._infer_root_cause(what) failure["root_cause"] = root_cause failure["five_whys"] = self._build_five_whys(what, root_cause) return failure def _step3_premortem(self, failure: Dict) -> Dict: """Step 3: Design the guardrail that prevents recurrence.""" what = failure.get("what", "").lower() root_cause = failure.get("root_cause", "").lower() guardrail = self._design_guardrail(what, root_cause) failure["guardrail_added"] = guardrail failure["never_again"] = ( f"Before {self._extract_action(what)}, always {guardrail.lower()}" ) return failure def _step4_evolve(self, failure: Dict): """Step 4: Write failure to all knowledge stores.""" # 4a: Failure log self._write_failure_log(failure) # 4b: KG entity self._write_kg_entity(failure) # 4c: Axiom axiom = self._generate_axiom(failure) self._write_axiom(axiom, failure) # 4d: Update TITAN MEMORY in CLAUDE.md (if high severity) if failure.get("severity", "low") in ("high", "critical"): self._update_titan_memory(failure) def _score_severity(self, failure: Dict) -> str: """Score failure severity: low / medium / high / critical.""" what = failure.get("what", "").lower() project = failure.get("project", "").lower() if any(domain in project for domain in HIGH_STAKES_DOMAINS): if any(kw in what for kw in ["data loss", "production down", "client", "revenue", "crash"]): return "critical" return "high" if any(kw in what for kw in ["error", "fail", "exception", "crash", "broken"]): return "medium" return "low" def _score_surprise(self, failure: Dict) -> float: """ Score how surprising this failure is (0.0 - 1.0). Higher = more surprising = more learning value. RLM workers activate on score >= SURPRISE_THRESHOLD (0.50). """ what = failure.get("what", "").lower() # Highly surprising: we thought something worked but it didn't if any(kw in what for kw in ["unexpected", "should have", "thought", "assumed", "worked before"]): return 0.85 # Medium surprise: known risk area we didn't guard if any(kw in what for kw in ["rate limit", "timeout", "auth", "permission", "404", "422"]): return 0.60 # Low surprise: routine errors return 0.35 def _infer_root_cause(self, what: str) -> str: """Rule-based root cause inference from failure description.""" patterns = [ (["rate limit", "429", "quota"], "API rate limits not pre-checked before execution"), (["auth", "401", "403", "permission", "unauthorized"], "Authentication state not verified before API call"), (["timeout", "connection refused", "network"], "Network/connectivity not validated before operation"), (["404", "not found", "missing"], "Resource existence not pre-verified"), (["422", "invalid", "validation"], "Input data not validated against API schema before submission"), (["import", "module not found", "no module"], "Dependency not installed or path not configured"), (["c: drive", "c:\\", "disk full", "space"], "C: drive used instead of E: drive — RULE 6 violation"), (["sqlite", ".db file", "sqlite3"], "SQLite used instead of Elestio storage — RULE 7 violation"), (["asked kinan", "manual step", "human needed"], "Automation not attempted before escalating — RULE 11 violation"), (["assumption", "assumed", "guessed"], "Decision made without research — RULE 8 (Research-First) violation"), ] for keywords, cause in patterns: if any(kw in what for kw in keywords): return cause return "Systemic gap not yet categorized — requires manual 5-Why analysis" def _build_five_whys(self, what: str, root_cause: str) -> List[str]: """Generate the 5-Why chain.""" return [ f"Why did this fail? — {what[:100]}", f"Why did that happen? — {root_cause}", "Why was that gap present? — No pre-execution check existed for this scenario", "Why was no check present? — Rule not yet codified or enforcement mechanism absent", "Why was the rule absent? — This failure pattern had not been encountered before" ] def _design_guardrail(self, what: str, root_cause: str) -> str: """Design a specific guardrail to prevent recurrence.""" guardrail_map = [ (["rate limit", "quota", "429"], "Check rate limit capacity before submitting batch requests"), (["auth", "401", "403"], "Verify authentication token is valid and non-expired before API calls"), (["timeout", "network"], "Implement retry with exponential backoff and circuit breaker"), (["not found", "404"], "Pre-verify resource existence with GET before PUT/PATCH/DELETE"), (["422", "validation"], "Validate payload against API schema before submission"), (["c: drive", "disk"], "Assert working directory starts with E: before any file operation"), (["sqlite"], "Grep for 'import sqlite3' before commit — fail build if found"), (["asked kinan", "manual"], "Run browser automation check before escalating to Kinan"), (["assumption", "assumed"], "Locate and read research files before writing any code"), ] for keywords, guardrail in guardrail_map: if any(kw in what or kw in root_cause for kw in keywords): return guardrail return "Add pre-execution verification step and automated test to detect this pattern" def _extract_action(self, what: str) -> str: """Extract the verb/action from failure description for the never_again rule.""" action_words = ["calling", "submitting", "writing", "reading", "executing", "running", "deploying"] for word in action_words: if word in what: return word return "executing" def _generate_axiom(self, failure: Dict) -> str: """Generate a single crisp axiom from the failure.""" guardrail = failure.get("guardrail_added", "verify before acting") project = failure.get("project", "general") return ( f"[{project.upper()}] Lesson: {guardrail}. " f"Root: {failure.get('root_cause', 'unknown')[:80]}. " f"Date: {failure.get('date')}." ) def _write_failure_log(self, failure: Dict): """Append failure to the master failure evolution log.""" with open(FAILURE_LOG_PATH, "a") as f: f.write(json.dumps(failure) + "\n") def _write_kg_entity(self, failure: Dict): """Write failure as a KG entity.""" today = datetime.now().strftime("%Y_%m_%d") entity_path = KG_ENTITIES_PATH / f"failure_evolution_{today}.jsonl" entity = { "entity_type": "failure_event", "entity_id": failure["id"], "timestamp": failure.get("timestamp"), "project": failure.get("project"), "task": failure.get("task"), "severity": failure.get("severity"), "surprise_score": failure.get("surprise_score"), "what": failure.get("what"), "root_cause": failure.get("root_cause"), "guardrail": failure.get("guardrail_added"), "never_again": failure.get("never_again"), "five_whys": failure.get("five_whys", []), "processed_by": "evolution_engine_v1" } with open(entity_path, "a") as f: f.write(json.dumps(entity) + "\n") def _write_axiom(self, axiom_text: str, failure: Dict): """Write axiom to the axioms directory.""" today = datetime.now().strftime("%Y_%m_%d") axiom_path = KG_AXIOMS_PATH / f"failure_axioms_{today}.jsonl" axiom_entry = { "axiom": axiom_text, "source": "evolution_engine", "failure_id": failure.get("id"), "project": failure.get("project"), "confidence": 0.85, "created_at": datetime.now().isoformat() } with open(axiom_path, "a") as f: f.write(json.dumps(axiom_entry) + "\n") self.axioms_generated += 1 def _update_titan_memory(self, failure: Dict): """ Update the TITAN MEMORY section in CLAUDE.md with high-severity learnings. Only called for high/critical failures. """ if not CLAUDE_MD_PATH.exists(): return learning_tag = failure.get("project", "system") learning_text = failure.get("guardrail_added", "verify before acting") date_str = failure.get("date", datetime.now().strftime("%Y-%m-%d")) new_learning = f"- **{learning_tag}**: {learning_text} ({date_str})" with open(CLAUDE_MD_PATH, "r", encoding="utf-8") as f: content = f.read() # Find the Titan Memory recent learnings section and insert target = "### Recent Learnings" if target in content: insert_pos = content.index(target) + len(target) + 1 # Find the first bullet point line lines = content[insert_pos:].split("\n") # Insert after the section header updated = ( content[:insert_pos] + "\n" + new_learning + "\n" + content[insert_pos:] ) with open(CLAUDE_MD_PATH, "w", encoding="utf-8") as f: f.write(updated) def scan_for_surprises(self, events: List[Dict]) -> List[Dict]: """ RLM Worker pattern: scan recent events for surprise patterns. Returns events with surprise_score >= SURPRISE_THRESHOLD. """ surprises = [] for event in events: score = self._score_surprise(event) if score >= SURPRISE_THRESHOLD: surprises.append({**event, "surprise_score": score}) return surprises def get_stats(self) -> Dict: """Return engine statistics.""" failure_count = 0 if FAILURE_LOG_PATH.exists(): with open(FAILURE_LOG_PATH, "r") as f: failure_count = sum(1 for line in f if line.strip()) return { "failures_processed_this_session": self.failures_processed, "axioms_generated_this_session": self.axioms_generated, "total_failures_in_log": failure_count, "surprise_threshold": SURPRISE_THRESHOLD, "high_stakes_domains": HIGH_STAKES_DOMAINS } if __name__ == "__main__": import argparse parser = argparse.ArgumentParser(description="Genesis Evolution Engine") parser.add_argument("--failure", type=str, help="Failure description to process") parser.add_argument("--project", type=str, default="general", help="Project ID") parser.add_argument("--task", type=str, default="manual", help="Task ID") parser.add_argument("--stats", action="store_true", help="Show engine stats") args = parser.parse_args() engine = EvolutionEngine() if args.stats: print(json.dumps(engine.get_stats(), indent=2)) elif args.failure: result = engine.process_failure({ "project": args.project, "task": args.task, "what": args.failure }) print(json.dumps(result, indent=2)) print(f"\nAxiom written. Genesis evolves.") else: print("Usage: python evolution_engine.py --failure 'description' [--project id] [--task id]") print(" python evolution_engine.py --stats")