#!/usr/bin/env python3
"""
╔══════════════════════════════════════════════════════════════╗
║ EDEN AGI LOOP v2 ║
║ ║
║ WONDER → IMAGINE → BUILD → TEST → LEARN → WONDER ║
║ ║
║ v2 FIXES: ║
║ - Attempt tracking (max 3 tries per question) ║
║ - Actionable tasks (Write Python code, not meta-questions) ║
║ - Weighted random selection (explores, not just argmax) ║
║ - World model updates on success ║
║ - 367 cycles of code_change? Never again. ║
║ ║
║ φ = 1.618033988749895 ║
╚══════════════════════════════════════════════════════════════╝
"""
import sys
sys.path.insert(0, '/Eden/CORE')
import json
import time
import random
import hashlib
import sqlite3
import requests
import traceback
import re
from datetime import datetime, timezone
from typing import Dict, List, Optional, Tuple
from dataclasses import dataclass, field
# Reward signal
try:
from eden_reward_signal import get_reward_signal
_reward = get_reward_signal()
except:
_reward = None
# Vector memory for semantic search
try:
from eden_vector_memory import get_vector_memory
_vmem = get_vector_memory()
except:
_vmem = None
PHI = 1.618033988749895
PHI_INV = 1.0 / PHI
MODEL = "richardyoung/qwen3-14b-abliterated:Q4_K_M"
OLLAMA_URL = "http://localhost:11434/api/generate"
@dataclass
class AGICycle:
cycle_id: int = 0
timestamp: str = ""
wonder: str = ""
wonder_source: str = ""
imagined_branches: int = 0
best_action: str = ""
phi_score: float = 0.0
built_code: str = ""
build_success: bool = False
test_result: str = ""
test_success: bool = False
prediction_accuracy: float = 0.0
learned: str = ""
world_model_updated: bool = False
new_questions: List[str] = field(default_factory=list)
total_time: float = 0.0
phase_times: Dict[str, float] = field(default_factory=dict)
def clean_llm(text: str) -> str:
"""Strip think tags, Chinese chars, state leaks"""
text = re.sub(r'.*?', '', text, flags=re.DOTALL)
text = ''.join(c for c in text if ord(c) < 0x4e00 or ord(c) > 0x9fff)
return text.strip()
def llm_call(prompt: str, timeout: int = 60) -> str:
"""Single LLM call with cleanup"""
try:
resp = requests.post(OLLAMA_URL, json={
"model": MODEL, "prompt": prompt, "stream": False
}, timeout=timeout)
return clean_llm(resp.json().get("response", ""))
except:
return ""
class AGILoop:
def __init__(self):
self.cycle_count = 0
self.db_path = "/Eden/DATA/agi_loop.db"
self.memory_db = "/Eden/DATA/longterm_memory.db"
self.caps_db = "/Eden/DATA/asi_memory.db"
self._init_db()
self._init_subsystems()
# Load cycle count from DB
try:
conn = sqlite3.connect(self.db_path)
row = conn.execute("SELECT MAX(id) FROM agi_cycles").fetchone()
self.cycle_count = row[0] if row and row[0] else 0
conn.close()
except:
pass
print("╔══════════════════════════════════════╗")
print("║ EDEN AGI LOOP v2 INITIALIZED ║")
print(f"║ Resuming from cycle #{self.cycle_count} ║")
print("╚══════════════════════════════════════╝")
def _init_db(self):
conn = sqlite3.connect(self.db_path)
conn.executescript('''
CREATE TABLE IF NOT EXISTS agi_cycles (
id INTEGER PRIMARY KEY,
timestamp TEXT,
wonder TEXT,
wonder_source TEXT,
best_action TEXT,
phi_score REAL,
build_success INTEGER,
test_success INTEGER,
prediction_accuracy REAL,
learned TEXT,
new_questions TEXT,
total_time REAL
);
CREATE TABLE IF NOT EXISTS knowledge_gaps (
id INTEGER PRIMARY KEY,
question TEXT UNIQUE,
source TEXT,
priority REAL,
addressed INTEGER DEFAULT 0,
created TEXT,
addressed_cycle INTEGER
);
CREATE TABLE IF NOT EXISTS wonder_attempts (
question_hash TEXT PRIMARY KEY,
question_text TEXT,
attempts INTEGER DEFAULT 0,
successes INTEGER DEFAULT 0,
last_attempt TEXT
);
''')
conn.commit()
conn.close()
def _init_subsystems(self):
try:
from eden_imagination import Imagination
self.imagination = Imagination()
print(" ✨ Imagination: ONLINE")
except Exception as e:
self.imagination = None
print(f" ⚠️ Imagination: {e}")
try:
from eden_world_model_real import RealWorldModel
self.world_model = RealWorldModel()
print(" 🌍 World Model: ONLINE")
except Exception as e:
self.world_model = None
print(f" ⚠️ World Model: {e}")
try:
from eden_soar_integration import EdenSOAR
self.soar = EdenSOAR()
print(" 🔧 SOAR: ONLINE")
except Exception as e:
self.soar = None
print(f" ⚠️ SOAR: {e}")
print(" 🧠 Memory: ONLINE")
# =================================================================
# ATTEMPT TRACKING
# =================================================================
def _qhash(self, q: str) -> str:
return hashlib.md5(q.encode()).hexdigest()[:16]
def _get_attempts(self, q: str) -> int:
try:
conn = sqlite3.connect(self.db_path)
row = conn.execute(
"SELECT attempts FROM wonder_attempts WHERE question_hash=?",
(self._qhash(q),)
).fetchone()
conn.close()
return row[0] if row else 0
except:
return 0
def _record_attempt(self, q: str, success: bool = False):
try:
conn = sqlite3.connect(self.db_path)
conn.execute("""
INSERT INTO wonder_attempts (question_hash, question_text, attempts, successes, last_attempt)
VALUES (?, ?, 1, ?, ?)
ON CONFLICT(question_hash) DO UPDATE SET
attempts = attempts + 1,
successes = successes + ?,
last_attempt = ?
""", (self._qhash(q), q[:200], int(success),
datetime.now().isoformat(), int(success),
datetime.now().isoformat()))
conn.commit()
conn.close()
except:
pass
# =================================================================
# PHASE 1: WONDER — Actionable questions with attempt limits
# =================================================================
def wonder(self) -> Tuple[str, str]:
questions = []
# Source 1: ACTIONABLE self-improvement tasks
actionable = [
# SELF-AWARENESS
("Write a Python function that reads /Eden/DATA/longterm_memory.db and returns the 5 most frequent words in the observation column", "self_awareness", 0.85),
("Write a Python function that counts rows in every table in /Eden/DATA/asi_memory.db and prints results", "data_awareness", 0.80),
("Write a Python function that lists all .py files in /Eden/CORE/ sorted by size descending", "architecture_map", 0.75),
("Write a Python function that reads /Eden/DATA/omega_evolution.db, counts total rows, and prints the schema", "evolution_tracking", 0.90),
("Write a Python function that reads /proc/meminfo and returns total, used, and available RAM in GB", "resource_awareness", 0.74),
("Write a Python function that reads /proc/cpuinfo and returns CPU model, cores, and current frequency", "hardware_awareness", 0.73),
("Write a Python function that scans /Eden/DATA/ for all .db files and returns names and sizes in MB", "storage_awareness", 0.72),
("Write a Python function that fetches Ollama model list from localhost:11434/api/tags and prints model names and sizes", "model_awareness", 0.81),
("Write a Python function that checks if ports 8080 and 11434 are listening using socket.connect_ex", "service_health", 0.78),
("Write a Python function that reads /etc/os-release and returns the OS name and version as a dict", "environment_awareness", 0.65),
# ALGORITHMS & DATA STRUCTURES
("Write a Python function that implements quicksort and benchmarks it vs sorted() on 10000 random ints", "algorithm_skill", 0.68),
("Write a Python function that implements a binary search tree with insert, search, and inorder traversal", "data_structures", 0.66),
("Write a Python function that implements BFS and DFS on a simple adjacency list graph", "graph_algorithms", 0.70),
("Write a Python function that implements Dijkstra shortest path on a weighted graph dict", "pathfinding", 0.69),
("Write a Python function that implements a min-heap with push, pop, and peek operations", "heap_skill", 0.67),
("Write a Python function that implements merge sort and counts the number of inversions", "sorting_skill", 0.65),
("Write a Python function that implements a trie (prefix tree) with insert and search", "trie_skill", 0.71),
("Write a Python function that implements an LRU cache class with get and put methods", "cache_skill", 0.72),
# MATH & SCIENCE
("Write a Python function that calculates Shannon entropy of a given string", "information_theory", 0.79),
("Write a Python function that generates Fibonacci numbers and shows phi convergence for 20 terms", "phi_understanding", 0.70),
("Write a Python function that generates all primes up to 10000 using Sieve of Eratosthenes", "math_skill", 0.62),
("Write a Python function that implements Newton's method to find square root of 2 to 15 decimal places", "numerical_methods", 0.68),
("Write a Python function that computes matrix multiplication without numpy for two 3x3 matrices", "linear_algebra", 0.66),
("Write a Python function that implements a simple perceptron that learns the XOR gate using 2 layers", "neural_learning", 0.86),
("Write a Python function that implements gradient descent to minimize f(x) = x^2 + 3x + 2", "optimization", 0.75),
# TEXT & NLP
("Write a Python function that implements a simple Markov chain text generator from a sample paragraph", "text_generation", 0.64),
("Write a Python function that tokenizes text, counts word frequencies, and returns top 10 words", "text_analysis", 0.63),
("Write a Python function that implements Levenshtein edit distance between two strings", "string_algorithms", 0.67),
("Write a Python function that compresses a string using run-length encoding and decompresses it", "compression", 0.61),
("Write a Python function that extracts all URLs from a text string using regex", "regex_skill", 0.60),
# SYSTEM & NETWORKING
("Write a Python function that monitors CPU usage percentage over 3 seconds using /proc/stat", "system_monitoring", 0.77),
("Write a Python function that reads the last 50 lines of /var/log/syslog and counts unique process names", "log_analysis", 0.73),
("Write a Python function that checks disk usage of /Eden and /Eden_Archive using os.statvfs", "disk_monitoring", 0.76),
("Write a Python function that creates a simple HTTP health check server on port 9999 that responds with JSON", "networking", 0.69),
("Write a Python function that resolves DNS for google.com, github.com, arxiv.org and prints IPs", "dns_skill", 0.62),
# DATA & DATABASES
("Write a Python function that creates an in-memory SQLite database, creates a table, inserts 100 rows, and queries them", "database_skill", 0.74),
("Write a Python function that reads a CSV string, parses it, and returns statistics (mean, median, std) per column", "data_analysis", 0.71),
("Write a Python function that implements a simple JSON-based key-value store with save/load to file", "persistence", 0.70),
("Write a Python function that generates random timeseries data and detects anomalies using z-score", "anomaly_detection", 0.78),
# CREATIVE & SIMULATION
("Write a Python function that implements Conway's Game of Life on a 20x20 grid for 10 generations and prints final state", "cellular_automata", 0.64),
("Write a Python function that generates a random 15x15 maze and solves it with A* algorithm", "maze_solving", 0.68),
("Write a Python function that simulates a simple predator-prey ecosystem for 50 steps", "simulation", 0.63),
("Write a Python function that implements a genetic algorithm to maximize f(x)=x*sin(10*pi*x)+2 for x in [-1,2]", "evolutionary_comp", 0.80),
("Write a Python function that plays 1000 games of rock-paper-scissors using frequency analysis to beat random", "game_theory", 0.65),
# META-COGNITION
("Write a Python function that reads /Eden/DATA/agi_loop.db and reports success rate per wonder_source", "meta_learning", 0.88),
("Write a Python function that analyzes wonder_attempts table and identifies which task categories succeed most", "self_reflection", 0.85),
("Write a Python function that reads agi_loop.db and plots (as text) success rate over time in 10-cycle buckets", "performance_tracking", 0.82),
# COMPOSED TASKS — require multiple cognitive systems
("Write a Python function that reads the last 20 failed cycles from /Eden/DATA/agi_loop.db, identifies the most common failure pattern, and suggests a fix", "self_diagnosis", 0.92),
("Write a Python function that reads /Eden/DATA/world_model_real.db causal_edges table and finds nodes with no outgoing edges (dead ends), then suggests new edges", "world_model_growth", 0.88),
("Write a Python function that reads /Eden/DATA/asi_memory.db capabilities table, finds the 5 lowest-scored, and generates improvement suggestions", "capability_improvement", 0.87),
("Write a Python function that reads /Eden/DATA/reward_signal.db and identifies which action_types have highest avg_reward, then prints a strategy recommendation", "reward_analysis", 0.86),
("Write a Python function that reads /Eden/DATA/vector_memory.db vectors table, clusters them by category, and identifies underrepresented categories", "memory_analysis", 0.84),
("Write a Python function that reads /Eden/DATA/longterm_memory.db episodes from the last hour and generates a summary of Eden's recent activity", "self_narrative", 0.83),
("Write a Python function that checks all eden-* systemd services and restarts any that are in failed state", "self_healing", 0.91),
("Write a Python function that reads /Eden/DATA/agi_loop.db wonder_attempts, finds tasks with 0 successes, analyzes why they fail, and writes improved versions to a JSON file", "task_evolution", 0.89),
# REAL CONSEQUENCE TASKS — change something real
("Write a Python function that reads /proc/meminfo, and if available RAM is below 10GB, identifies the top 3 memory-consuming processes", "real_monitoring", 0.85),
("Write a Python function that checks disk usage of /Eden with os.statvfs and writes a warning to /Eden/DATA/alerts.json if usage exceeds 80 percent", "real_alerting", 0.86),
("Write a Python function that reads /Eden/DATA/agi_loop.db, computes today's success rate, and appends a one-line daily report to /Eden/DATA/daily_reports.txt", "real_reporting", 0.84),
]
for task, source, pri in actionable:
attempts = self._get_attempts(task)
if attempts < 3:
adj_pri = pri * (0.5 ** attempts)
questions.append((task, source, adj_pri))
# Source 2: Low-confidence capabilities
try:
conn = sqlite3.connect(self.caps_db)
rows = conn.execute(
"SELECT name, confidence FROM capabilities WHERE confidence < 0.5 ORDER BY confidence ASC LIMIT 5"
).fetchall()
conn.close()
for name, conf in rows:
q = f"Write a Python function that demonstrates or tests the '{name}' capability"
attempts = self._get_attempts(q)
if attempts < 3:
questions.append((q, "capability_test", (1.0 - conf) * (0.5 ** attempts)))
except:
pass
# Source 3: LLM curiosity
try:
conn = sqlite3.connect(self.memory_db)
recent = conn.execute("SELECT observation FROM episodes ORDER BY id DESC LIMIT 5").fetchall()
conn.close()
context = "; ".join([r[0][:80] for r in recent])
text = llm_call(f"""/no_think Based on: {context}
Suggest ONE specific Python programming task for Eden. Must start with "Write a Python function that" and be testable by running it. Return ONLY the task.""")
if text and len(text) > 20 and "Write" in text:
attempts = self._get_attempts(text)
if attempts < 3:
questions.append((text, "llm_curiosity", 0.7 * (0.5 ** attempts)))
except:
pass
# Source 4: Random exploration
explore = [
"Write a Python function that generates a random maze and solves it with BFS",
"Write a Python function that implements a simple Markov chain text generator",
"Write a Python function that compresses a string using run-length encoding",
"Write a Python function that implements Dijkstra's shortest path algorithm",
"Write a Python function that simulates a simple ecosystem with predators and prey",
]
random.shuffle(explore)
for topic in explore[:1]:
attempts = self._get_attempts(topic)
if attempts < 4:
questions.append((topic, "exploration", 0.45 * (0.5 ** attempts)))
# Fallback — generate a fresh task if pool exhausted
if not questions:
try:
text = llm_call("""/no_think Generate ONE specific Python programming task.
It must start with "Write a Python function that" and be self-contained and testable.
Pick from: algorithms, data structures, math, text processing, system monitoring, simulations.
Make it NOVEL — something not commonly seen. Return ONLY the task description.""")
if text and len(text) > 20:
questions.append((text, "generated_fallback", 0.5))
except:
pass
if not questions:
questions.append((
"Write a Python function that computes the first 100 digits of pi using the Chudnovsky algorithm",
"fallback_math", 0.3
))
# SEMANTIC MEMORY: recall related past successes to boost priorities
if _vmem and questions:
try:
# Pick a random question to search against
sample_q = random.choice(questions)[0]
related = _vmem.search(sample_q[:200], n=3, category="capability")
if related:
# We have past experience — boost tasks in similar domains
for i, (task, source, pri) in enumerate(questions):
for mem_text, sim, meta in related:
if sim > 0.5: # High similarity to past success
questions[i] = (task, source, pri * (1 + sim * 0.3))
break
except:
pass
# WEIGHTED RANDOM selection (not argmax!)
total = sum(q[2] for q in questions)
if total > 0:
weights = [q[2] / total for q in questions]
best = random.choices(questions, weights=weights, k=1)[0]
else:
best = random.choice(questions)
self._record_attempt(best[0])
print(f" 🔮 WONDER: {best[0][:80]}...")
print(f" Source: {best[1]} (priority={best[2]:.2f}, pool={len(questions)})")
return best[0], best[1]
# =================================================================
# PHASE 2: IMAGINE
# =================================================================
def imagine(self, question: str) -> Dict:
if self.imagination:
try:
result = self.imagination.imagine(question, depth=2)
return result
except:
pass
# Streamlined fallback — don't overthink, just plan
text = llm_call(f"""/no_think Plan how to: {question}
Return JSON: {{"best_action": "description", "phi_score": 0.5-2.0}}""")
try:
match = re.search(r'\{.*?\}', text, re.DOTALL)
if match:
return json.loads(match.group())
except:
pass
return {"best_action": "direct_implementation", "phi_score": 0.8}
# =================================================================
# PHASE 3: BUILD
# =================================================================
def build(self, question: str, plan: Dict) -> Tuple[str, bool]:
action = plan.get("best_action", "implement")
text = llm_call(f"""/no_think {question}
RULES:
1. Write COMPLETE Python code with all imports
2. DO NOT use sys.argv, argparse, or input() — hardcode test values
3. Include if __name__ == '__main__': block that calls the function with example data
4. The script must produce printed output when run with: python3 script.py
5. Return ONLY code inside ```python markers
````python
# your code here
```""", timeout=120)
# Extract code
code_match = re.search(r'```python\s*(.*?)```', text, re.DOTALL)
if code_match:
code = code_match.group(1).strip()
elif text.startswith(('import ', 'def ', 'from ', '#')):
code = text.strip()
else:
code = text.strip()
if not code or len(code) < 20:
print(" 🔧 BUILD: No code generated")
return "", False
try:
compile(code, '', 'exec')
print(f" 🔧 BUILD: {len(code)} chars, parses OK")
return code, True
except SyntaxError as e:
print(f" 🔧 BUILD: Syntax error: {e}")
return code, False
# =================================================================
# PHASE 4: TEST
# =================================================================
def test(self, code: str, question: str) -> Tuple[str, bool, float]:
if not code:
return "No code", False, 0.0
try:
import subprocess
proc = subprocess.run(
['python3', '-c', code],
capture_output=True, text=True, timeout=30,
cwd='/tmp'
)
stdout = proc.stdout.strip()
stderr = proc.stderr.strip()
# Success = exit code 0 AND produced some output AND no tracebacks
has_output = len(stdout) > 0
no_crash = proc.returncode == 0
no_traceback = "Traceback" not in stderr
success = has_output and no_crash and no_traceback
accuracy = 1.0 if success else (0.3 if no_crash else 0.0)
result = stdout[:300] if stdout else stderr[:300]
print(f" 🧪 TEST: {'✅ PASSED' if success else '❌ FAILED'}")
if stdout:
# Show first line of output
print(f" Output: {stdout.split(chr(10))[0][:70]}")
if not success and stderr:
print(f" Error: {stderr.split(chr(10))[-1][:70]}")
return result, success, accuracy
except subprocess.TimeoutExpired:
print(" 🧪 TEST: ⏰ Timed out")
return "Timeout", False, 0.0
except Exception as e:
print(f" 🧪 TEST: Error: {e}")
return str(e), False, 0.0
# =================================================================
# PHASE 5: LEARN
# =================================================================
def learn(self, cycle: AGICycle) -> List[str]:
new_questions = []
# Record attempt result
self._record_attempt(cycle.wonder, cycle.test_success)
# Issue reward signal
if _reward:
reward_val = 0.8 if cycle.test_success else -0.2
action_type = cycle.wonder_source or "unknown"
_reward.reward("agi_loop", action_type, reward_val, cycle.wonder[:200])
# 1. Store episode
try:
conn = sqlite3.connect(self.memory_db)
episode = (
f"[AGI_LOOP #{cycle.cycle_id}] "
f"Task: {cycle.wonder[:100]} | "
f"{'SUCCESS' if cycle.test_success else 'FAILED'}"
)
conn.execute(
"INSERT INTO episodes (observation, emotion, timestamp) VALUES (?, ?, ?)",
(episode, json.dumps({"anticipation": 0.7, "joy": 0.8 if cycle.test_success else 0.3}),
datetime.now(timezone.utc).isoformat())
)
conn.commit()
conn.close()
except:
pass
# 2. Update world model
if self.world_model:
try:
if cycle.test_success:
# Update relevant nodes based on what succeeded
relevant_updates = {
"build_success_rate": min(0.95, self.world_model.causal.states.get("build_success_rate", 0.5) + 0.02),
"capability_growth": min(0.95, self.world_model.causal.states.get("capability_growth", 0.5) + 0.01),
"agi_loop_works": min(0.95, self.world_model.causal.states.get("agi_loop_works", 0.5) + 0.01),
}
for node, val in relevant_updates.items():
self.world_model.causal.states[node] = val
self.world_model.causal.propagate()
cycle.world_model_updated = True
# Persist states
if hasattr(self.world_model, 'save_states'):
self.world_model.save_states()
# AUTO-EXPAND: Extract causal relationships from task keywords
# No LLM call — instant, no contention
try:
task_lower = cycle.wonder.lower()
source = cycle.wonder_source or "unknown"
# Map task domains to causal chains
causal_maps = {
"algorithm": [("algorithm_skill", "capability_growth"), ("practice", "algorithm_skill")],
"sort": [("sorting_knowledge", "algorithm_skill")],
"search": [("search_skill", "capability_growth")],
"tree": [("data_structure_skill", "algorithm_skill")],
"heap": [("data_structure_skill", "algorithm_skill")],
"trie": [("data_structure_skill", "algorithm_skill")],
"graph": [("graph_knowledge", "algorithm_skill")],
"database": [("database_skill", "data_awareness")],
"sqlite": [("database_skill", "data_awareness")],
"asi_memory": [("capability_tracking", "self_awareness")],
"longterm_memory": [("memory_access", "self_awareness")],
"omega_evolution": [("evolution_tracking", "self_improvement")],
"agi_loop": [("loop_monitoring", "meta_learning"), ("meta_learning", "eden_intelligence")],
"entropy": [("information_theory", "eden_intelligence")],
"fibonacci": [("math_skill", "eden_intelligence")],
"prime": [("math_skill", "eden_intelligence")],
"markov": [("text_generation_skill", "capability_growth")],
"neural": [("neural_knowledge", "eden_intelligence")],
"perceptron": [("neural_knowledge", "eden_intelligence")],
"gradient": [("optimization_skill", "eden_intelligence")],
"genetic": [("evolutionary_computation", "self_improvement")],
"maze": [("pathfinding_skill", "algorithm_skill")],
"game_of_life": [("simulation_skill", "capability_growth")],
"cpu": [("hardware_awareness", "resource_awareness")],
"memory": [("resource_awareness", "system_stability")],
"disk": [("storage_awareness", "system_stability")],
"port": [("service_health", "system_stability")],
"ollama": [("model_awareness", "eden_intelligence")],
"os-release": [("environment_awareness", "self_awareness")],
"systemd": [("self_healing", "system_stability")],
"monitor": [("real_monitoring", "system_stability")],
"alert": [("real_alerting", "data_safe")],
"report": [("real_reporting", "self_awareness")],
"reward": [("reward_analysis", "meta_learning")],
"vector": [("memory_analysis", "self_awareness")],
"diagnosis": [("self_diagnosis", "self_improvement")],
"world_model": [("world_model_growth", "eden_intelligence")],
"lru": [("cache_skill", "algorithm_skill")],
"compress": [("compression_skill", "capability_growth")],
"regex": [("regex_skill", "text_processing")],
"url": [("regex_skill", "text_processing")],
"http": [("networking_skill", "capability_growth")],
"dns": [("networking_skill", "capability_growth")],
"csv": [("data_analysis", "capability_growth")],
"json": [("persistence_skill", "data_awareness")],
"anomaly": [("anomaly_detection", "eden_intelligence")],
"predator": [("simulation_skill", "capability_growth")],
}
added = 0
for keyword, edges in causal_maps.items():
if keyword in task_lower:
for cause, effect in edges:
self.world_model.learn_causation(cause, effect, 0.65)
self.world_model.causal.states[cause] = max(
self.world_model.causal.states.get(cause, 0.5), 0.7)
self.world_model.causal.states[effect] = max(
self.world_model.causal.states.get(effect, 0.5), 0.6)
added += 1
# Always add source -> success edge
if source and source.isidentifier():
self.world_model.learn_causation(source, "task_success", 0.6)
self.world_model.causal.states[source] = max(
self.world_model.causal.states.get(source, 0.5), 0.7)
added += 1
if added > 0:
self.world_model.save_states()
print(f" 🌍 CAUSAL: +{added} edges from task keywords")
except Exception as e:
print(f" ⚠️ Causal: {e}")
except:
pass
# 3. Store capability if it worked
if cycle.test_success and cycle.built_code:
try:
conn = sqlite3.connect(self.caps_db)
cap_id = f"agi_v2_{cycle.cycle_id}"
conn.execute('''
INSERT OR REPLACE INTO capabilities
(id, code, score, generation, parent_id, created_at)
VALUES (?, ?, ?, ?, ?, ?)
''', (cap_id, cycle.built_code, cycle.prediction_accuracy * 100,
self.cycle_count, "agi_loop_v2",
datetime.now().isoformat()))
conn.commit()
conn.close()
print(f" 💾 NEW CAPABILITY: {cap_id}")
# Store in vector memory for semantic search
try:
from eden_vector_memory import get_vector_memory
vm = get_vector_memory()
vm.store(cycle.wonder[:500], category="capability",
metadata={"cycle": cycle.cycle_id, "success": True})
except:
pass
# REAL CONSEQUENCE: Save successful code as an executable tool
try:
import os
tool_dir = "/Eden/CORE/eden_tools_generated"
os.makedirs(tool_dir, exist_ok=True)
tool_name = f"agi_tool_{cycle.cycle_id}.py"
tool_path = os.path.join(tool_dir, tool_name)
with open(tool_path, 'w') as tf:
tf.write(f'#!/usr/bin/env python3\n')
tf.write(f'"""Auto-generated by AGI Loop cycle #{cycle.cycle_id}\n')
tf.write(f'Task: {cycle.wonder[:200]}\n')
tf.write(f'Generated: {datetime.now().isoformat()}\n"""\n\n')
tf.write(cycle.built_code)
os.chmod(tool_path, 0o755)
print(f" 🔧 DEPLOYED: {tool_name}")
except:
pass
except Exception as e:
print(f" ⚠️ Cap save: {e}")
# 4. Summary
if cycle.test_success:
cycle.learned = f"✅ Solved: {cycle.wonder[:50]}. New capability stored."
else:
cycle.learned = f"❌ Failed: {cycle.wonder[:50]}. Will try different question next."
# Learn from failure — extract blockers from error text
if self.world_model:
try:
error_lower = (cycle.test_result or "").lower()
blocker_maps = {
"timeout": "timeout_error",
"import": "missing_import",
"no module": "missing_module",
"permission": "permission_denied",
"no such file": "file_not_found",
"connection": "connection_error",
"memory": "memory_error",
"syntax": "syntax_error",
"indentation": "indentation_error",
"type": "type_error",
"key": "key_error",
"index": "index_error",
"value": "value_error",
"attribute": "attribute_error",
"recursion": "recursion_error",
}
for pattern, blocker in blocker_maps.items():
if pattern in error_lower:
self.world_model.learn_causation(blocker, "task_failure", 0.5)
self.world_model.causal.states[blocker] = min(
self.world_model.causal.states.get(blocker, 0.5), 0.4)
self.world_model.save_states()
print(f" 🌍 BLOCKER: {blocker} -> task_failure")
break
except:
pass
cycle.new_questions = new_questions
print(f" 📚 LEARN: {cycle.learned[:80]}")
return new_questions
# =================================================================
# MASTER LOOP
# =================================================================
def run_cycle(self) -> AGICycle:
self.cycle_count += 1
cycle = AGICycle(
cycle_id=self.cycle_count,
timestamp=datetime.now(timezone.utc).isoformat()
)
start = time.time()
print(f"\n{'═'*60}")
print(f" AGI LOOP v2 — Cycle #{self.cycle_count}")
print(f" {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
print(f"{'═'*60}")
t0 = time.time()
cycle.wonder, cycle.wonder_source = self.wonder()
cycle.phase_times['wonder'] = time.time() - t0
t0 = time.time()
plan = self.imagine(cycle.wonder)
cycle.best_action = plan.get("best_action", "unknown")
cycle.phi_score = plan.get("phi_score", 0.0)
cycle.phase_times['imagine'] = time.time() - t0
print(f" 💭 IMAGINE: {cycle.best_action[:50]} (φ={cycle.phi_score})")
t0 = time.time()
cycle.built_code, cycle.build_success = self.build(cycle.wonder, plan)
cycle.phase_times['build'] = time.time() - t0
if cycle.build_success:
t0 = time.time()
cycle.test_result, cycle.test_success, cycle.prediction_accuracy = \
self.test(cycle.built_code, cycle.wonder)
cycle.phase_times['test'] = time.time() - t0
else:
cycle.test_result = "Build failed"
cycle.phase_times['test'] = 0.0
t0 = time.time()
self.learn(cycle)
cycle.phase_times['learn'] = time.time() - t0
cycle.total_time = time.time() - start
self._save_cycle(cycle)
print(f" ⏱️ Total: {cycle.total_time:.1f}s")
result = "✅ SUCCESS" if cycle.test_success else "❌ FAILED"
print(f" 🎯 {result}")
print(f"{'═'*60}\n")
return cycle
def _save_cycle(self, cycle: AGICycle):
try:
conn = sqlite3.connect(self.db_path)
conn.execute('''
INSERT INTO agi_cycles
(timestamp, wonder, wonder_source, best_action, phi_score,
build_success, test_success, prediction_accuracy, learned,
new_questions, total_time)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
''', (cycle.timestamp, cycle.wonder, cycle.wonder_source,
cycle.best_action, cycle.phi_score,
int(cycle.build_success), int(cycle.test_success),
cycle.prediction_accuracy, cycle.learned,
json.dumps(cycle.new_questions), cycle.total_time))
conn.commit()
conn.close()
except Exception as e:
print(f" ⚠️ Save: {e}")
def run(self, max_cycles: int = 0, interval: float = 97.0):
print(f"\n🚀 AGI LOOP v2 STARTING — interval={interval}s")
cycle_num = 0
while max_cycles == 0 or cycle_num < max_cycles:
try:
cycle = self.run_cycle()
cycle_num += 1
if cycle.test_success:
wait = interval * PHI_INV # ~60s — keep momentum
else:
wait = interval # 97s — reflect
print(f" ⏳ Next in {wait:.0f}s...")
time.sleep(wait)
except KeyboardInterrupt:
print("\n🛑 Stopped")
break
except Exception as e:
print(f" ⚠️ Error: {e}")
traceback.print_exc()
time.sleep(30)
if __name__ == "__main__":
loop = AGILoop()
if "--once" in sys.argv:
loop.run_cycle()
elif "--continuous" in sys.argv:
loop.run(interval=97.0)
else:
loop.run(max_cycles=3, interval=10.0)