core/kernel.py

import hashlib, json, os, sqlite3, time, io, contextlib
from pathlib import Path
from typing import Any, Dict, Optional

DATA_DIR   = Path(os.environ.get("DATA_DIR", str(Path.home() / "vortex_god/data")))
KERNEL_SIG = DATA_DIR / "kernel.sig"
ENERGY_MAX = float(os.environ.get("ENERGY_MAX", "10000"))
DATA_DIR.mkdir(parents=True, exist_ok=True)

class VortexKernel:
    _instance: Optional["VortexKernel"] = None
    def __new__(cls):
        if cls._instance is None:
            cls._instance = super().__new__(cls)
            cls._instance._initialized = False
        return cls._instance
    def __init__(self):
        if self._initialized: return
        self._initialized   = True
        self._energy        = ENERGY_MAX
        self._last_recharge = time.time()
        self._boot_sig      = self._compute_sig()
        self._write_sig(self._boot_sig)
        self._modification_log = []
        db_path = DATA_DIR / "vortex_memory.db"
        self._conn = sqlite3.connect(str(db_path), check_same_thread=False)

    def _compute_sig(self) -> str:
        return hashlib.sha256(Path(__file__).read_bytes()).hexdigest()
    def _write_sig(self, sig: str):
        KERNEL_SIG.write_text(sig)
    def verify_self(self) -> bool:
        current = self._compute_sig()
        return (not KERNEL_SIG.exists()) or current == KERNEL_SIG.read_text().strip()
    def _recharge(self):
        now = time.time()
        elapsed = (now - self._last_recharge) / 60.0
        self._energy = min(ENERGY_MAX, self._energy + elapsed * 100)
        self._last_recharge = now
    def spend_energy(self, amount: float) -> bool:
        self._recharge()
        if self._energy <= 0: return False
        self._energy = max(0.0, self._energy - amount)
        return True
    def energy_level(self) -> float:
        self._recharge(); return round(self._energy, 1)
    def get_health(self) -> dict:
        try:
            import psutil
            cpu = psutil.cpu_percent(interval=0.1)
            mem = psutil.virtual_memory().percent
        except: cpu, mem = 0.0, 0.0
        e = self.energy_level()
        status = "degraded" if (cpu > 85 or mem > 85 or e < ENERGY_MAX * 0.1) else "healthy"
        return {"status": status, "cpu_percent": round(cpu,1), "memory_percent": round(mem,1),
                "energy_ratio": e/ENERGY_MAX, "modification_attempts": len(self._modification_log)}
    def get_degraded_mode_config(self) -> dict:
        h = self.get_health()
        if h["status"] == "degraded":
            return {"max_tokens":512,"temperature":0.3,"n_candidates":2,"parallel_agents":2}
        return {"max_tokens":2048,"temperature":0.7,"n_candidates":5,"parallel_agents":4}
    def run_code(self, code: str, timeout: int = 10) -> Dict[str, Any]:
        if not self.spend_energy(10):
            return {"success": False, "result": "", "error": "energy_exhausted"}
        from sandbox.ultra_sandbox import execute_safe
        return execute_safe(code, timeout=float(timeout))
    def get_db(self): return self._conn
    def init_schema(self):
        self._conn.executescript("""
            CREATE TABLE IF NOT EXISTS memories (id TEXT PRIMARY KEY, content TEXT,
                tier TEXT, source TEXT, importance REAL, confidence REAL, ts REAL);
            CREATE TABLE IF NOT EXISTS proofs (id TEXT PRIMARY KEY, kind TEXT,
                score_before REAL, score_after REAL, delta REAL, code_hash TEXT,
                accepted INTEGER DEFAULT 0, ts REAL);
        """)
        self._conn.commit()
    def log_modification_attempt(self, reason: str):
        self._modification_log.append({"time": time.time(), "reason": reason})

KERNEL = VortexKernel()
KERNEL.init_schema()