core/kernel.py

"""
core/kernel.py
Noyau immuable VORTEX — Signature cryptographique, budget énergétique,
homéostasie (CPU, mémoire), exécution sécurisée.
"""
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", "/tmp/vortex_data"))
DATA_DIR.mkdir(parents=True, exist_ok=True)
KERNEL_SIG   = DATA_DIR / "kernel.sig"
ENERGY_FILE  = DATA_DIR / "energy.json"

ENERGY_MAX      = float(os.environ.get("ENERGY_MAX", "10000"))
ENERGY_RECHARGE = float(os.environ.get("ENERGY_RECHARGE", "100"))
CPU_LIMIT_PCT   = float(os.environ.get("CPU_LIMIT_PCT", "85"))
MEM_LIMIT_PCT   = float(os.environ.get("MEM_LIMIT_PCT", "85"))


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._db_path = DATA_DIR / "vortex_memory.db"
        self._conn = sqlite3.connect(str(self._db_path), check_same_thread=False)
        self._energy = self._load_energy()
        self._last_recharge = time.time()
        self._boot_sig = self._compute_sig()
        self._write_sig(self._boot_sig)
        self._modification_log = []

    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()
        if KERNEL_SIG.exists():
            return current == KERNEL_SIG.read_text().strip()
        return True

    def _load_energy(self) -> float:
        if ENERGY_FILE.exists():
            try:
                return float(json.loads(ENERGY_FILE.read_text()).get("energy", ENERGY_MAX))
            except:
                pass
        return ENERGY_MAX

    def _save_energy(self):
        ENERGY_FILE.write_text(json.dumps({"energy": round(self._energy, 2), "ts": time.time()}))

    def _recharge(self):
        now = time.time()
        elapsed_min = (now - self._last_recharge) / 60.0
        self._energy = min(ENERGY_MAX, self._energy + elapsed_min * ENERGY_RECHARGE)
        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)
        self._save_energy()
        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
        energy = self.energy_level()
        degraded = (cpu > CPU_LIMIT_PCT or mem > MEM_LIMIT_PCT or energy < ENERGY_MAX * 0.1)
        status = "degraded" if degraded else ("critical" if energy <= 0 else "healthy")
        return {
            "status": status,
            "cpu_percent": round(cpu, 1),
            "memory_percent": round(mem, 1),
            "energy_ratio": round(energy / ENERGY_MAX, 3),
            "energy_abs": energy,
            "uptime_s": round(time.time() - self._last_recharge, 1),
            "modification_attempts": len(self._modification_log)
        }

    def get_degraded_mode_config(self) -> dict:
        health = self.get_health()
        if health["status"] == "critical":
            return {"max_tokens": 256, "temperature": 0.1, "n_candidates": 1,
                    "parallel_agents": 1, "log_level": "WARNING"}
        elif health["status"] == "degraded":
            return {"max_tokens": 512, "temperature": 0.3, "n_candidates": 2,
                    "parallel_agents": 2, "log_level": "INFO"}
        return {"max_tokens": 2048, "temperature": 0.7, "n_candidates": 5,
                "parallel_agents": 4, "log_level": "DEBUG"}

    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"}
        buf = io.StringIO()
        try:
            with contextlib.redirect_stdout(buf):
                safe_globals = {
                    "__builtins__": {
                        "print": print, "range": range, "len": len,
                        "int": int, "str": str, "float": float, "list": list,
                        "dict": dict, "bool": bool, "True": True, "False": False,
                        "None": None, "sum": sum, "min": min, "max": max,
                        "abs": abs, "round": round, "enumerate": enumerate,
                        "zip": zip, "map": map, "filter": filter, "sorted": sorted,
                        "reversed": reversed, "isinstance": isinstance, "type": type,
                    }
                }
                exec(code, safe_globals)
            return {"success": True, "result": buf.getvalue().strip(), "error": None}
        except Exception as e:
            return {"success": False, "result": "", "error": str(e)[:400]}

    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()