core/auto_evolution.py

# core/auto_evolution.py
"""
Auto-évolution améliorée – propositions proactives même sans lacunes.
"""
import time
import logging
import threading
import asyncio
from typing import Dict, List, Optional
from dataclasses import dataclass

logger = logging.getLogger("lucie.auto_evolution")

@dataclass
class EvolutionProposal:
    id: str
    type: str
    description: str
    code: str
    score: float
    status: str = "pending"

class AutoEvolution:
    def __init__(self, hermes_loop, fable_engine, error_tree, vector_memory,
                 skill_suite_prompting, validation_queue, chat_queue=None):
        self.hermes = hermes_loop
        self.fable = fable_engine
        self.error_tree = error_tree
        self.vector_memory = vector_memory
        self.skill_suite = skill_suite_prompting
        self.validation_queue = validation_queue
        self.chat_queue = chat_queue or []
        self.proposals: Dict[str, EvolutionProposal] = {}
        self.is_running = False
        self.last_activity = time.time()
        self.inactivity_threshold = 180  # 3 min
        self.cooldown = 600  # 10 min entre deux analyses
        self.last_scan_result = None
        self.last_train_time = 0
        self.train_interval = 3600 * 24  # 24h entre deux auto‑train

    def analyze_gaps(self) -> List[Dict]:
        """Analyse les lacunes actuelles (scores, erreurs)."""
        gaps = []
        try:
            recent = self.vector_memory.search("*", n=20, min_score=0.0)
            scores = [ep.get("metadata", {}).get("score", 0.5) for ep in recent if ep.get("metadata", {}).get("score")]
            if scores:
                avg_score = sum(scores) / len(scores)
                if avg_score < 0.7:
                    gaps.append({
                        "type": "performance",
                        "description": f"Score moyen bas ({avg_score:.2f}).",
                        "severity": "high",
                        "suggestion": "Optimiser les cycles Hermès/Fable"
                    })
            if hasattr(self.error_tree, "get_recent_errors"):
                errors = self.error_tree.get_recent_errors(limit=10)
                error_types = {}
                for e in errors:
                    typ = e.get("type", "unknown")
                    error_types[typ] = error_types.get(typ, 0) + 1
                for typ, count in error_types.items():
                    if count >= 2:
                        gaps.append({
                            "type": "error_pattern",
                            "description": f"Erreur '{typ}' répétée {count} fois.",
                            "severity": "medium",
                            "suggestion": f"Créer une compétence pour gérer '{typ}'"
                        })
            if gaps:
                for gap in gaps:
                    keywords = gap["description"].split()
                    for kw in keywords[:3]:
                        if len(kw) > 4:
                            skills = self.vector_memory.search_skills(kw, n=1, min_score=0.3)
                            if not skills:
                                gaps.append({
                                    "type": "missing_skill",
                                    "description": f"Aucune compétence pour '{kw}'.",
                                    "severity": "medium",
                                    "suggestion": f"Générer une compétence '{kw}'"
                                })
                                break
        except Exception as e:
            logger.error(f"Erreur auto-diagnostic : {e}")
        return gaps

    def _find_opportunities(self) -> List[Dict]:
        """Trouve des opportunités d'amélioration même sans lacunes."""
        opportunities = []
        # 1. Temps depuis le dernier train
        if time.time() - self.last_train_time > self.train_interval:
            opportunities.append({
                "type": "opportunity",
                "description": "Plus de 24h sans fine‑tuning.",
                "severity": "low",
                "suggestion": "Lancer un nouveau cycle @train pour s'améliorer."
            })
        # 2. Peu d'épisodes en mémoire
        stats = self.vector_memory.stats() if self.vector_memory else {}
        count = stats.get("count", 0)
        if count < 50:
            opportunities.append({
                "type": "opportunity",
                "description": f"Seulement {count} épisodes en mémoire.",
                "severity": "low",
                "suggestion": "Explorer de nouveaux sujets pour enrichir la base."
            })
        # 3. Proposer une compétence générique
        opportunities.append({
            "type": "opportunity",
            "description": "Générer une compétence d'optimisation de code.",
            "severity": "low",
            "suggestion": "Créer une compétence pour améliorer les performances."
        })
        return opportunities

    async def generate_skill(self, task: str) -> Optional[Dict]:
        try:
            result = await self.fable.run(
                task,
                max_iter=3,
                feedback="Génère une compétence Python autonome, avec documentation et tests."
            )
            if result.get("best_score", 0) >= 0.8:
                code = self._extract_code(result.get("best_result", ""))
                return {
                    "description": task,
                    "code": code,
                    "score": result["best_score"],
                    "type": "new_skill"
                }
        except Exception as e:
            logger.error(f"Erreur génération skill : {e}")
        return None

    def _extract_code(self, text: str) -> str:
        import re
        match = re.search(r'```python\s*(.*?)\s*```', text, re.DOTALL)
        if match:
            return match.group(1)
        return text

    async def propose_skill(self, task: str) -> Optional[EvolutionProposal]:
        skill_data = await self.generate_skill(task)
        if not skill_data:
            return None
        proposal = EvolutionProposal(
            id=f"prop_{int(time.time())}",
            type="new_skill",
            description=skill_data["description"],
            code=skill_data["code"],
            score=skill_data["score"]
        )
        self.proposals[proposal.id] = proposal
        return proposal

    def start_consciousness(self):
        if self.is_running:
            return
        self.is_running = True
        threading.Thread(target=self._consciousness_loop, daemon=True).start()
        logger.info("🧠 Mode conscience de soi activé.")

    def _consciousness_loop(self):
        while self.is_running:
            time.sleep(60)  # Vérification toutes les minutes
            if not self._is_inactive():
                continue
            if self.last_scan_result and time.time() - self.last_scan_result < self.cooldown:
                continue

            logger.info("🔍 Silence détecté – analyse en cours...")
            gaps = self.analyze_gaps()
            self.last_scan_result = time.time()

            if gaps:
                # Priorité aux lacunes
                gap = max(gaps, key=lambda x: {"high": 3, "medium": 2, "low": 1}.get(x["severity"], 0))
                proposal = asyncio.run(self.propose_skill(gap["suggestion"]))
            else:
                # Pas de lacune → chercher des opportunités
                opportunities = self._find_opportunities()
                if opportunities:
                    # Prendre la première opportunité
                    opp = opportunities[0]
                    proposal = asyncio.run(self.propose_skill(opp["suggestion"]))
                else:
                    proposal = None

            if proposal:
                self._notify_user(proposal)
                # Mettre à jour le timestamp du dernier train si c'était un @train
                if "train" in proposal.description.lower():
                    self.last_train_time = time.time()

    def _is_inactive(self) -> bool:
        return (time.time() - self.last_activity) > self.inactivity_threshold

    def _notify_user(self, proposal: EvolutionProposal):
        msg = (
            f"🧠 **LUCIE propose une amélioration**\n\n"
            f"**Type** : {proposal.type}\n"
            f"**Description** : {proposal.description}\n"
            f"**Score** : {proposal.score:.2f}\n\n"
            f"Code généré :\n```python\n{proposal.code[:500]}\n```\n\n"
            f"✅ Valider : `@approve {proposal.id}`\n"
            f"❌ Refuser : `@reject {proposal.id}`"
        )
        if self.chat_queue is not None:
            self.chat_queue.append({"role": "assistant", "content": msg})

    def approve_proposal(self, proposal_id: str) -> bool:
        prop = self.proposals.get(proposal_id)
        if not prop or prop.status != "pending":
            return False
        try:
            from core.skill_schema import Skill, SkillMetadata
            skill = Skill(
                metadata=SkillMetadata(
                    name=f"auto_{int(time.time())}",
                    description=prop.description[:100],
                    category="auto_generated"
                ),
                code=prop.code,
                score=prop.score
            )
            self.vector_memory.add_skill(skill)
            import os
            skill_dir = f"/data/skills/{skill.metadata.name}"
            os.makedirs(skill_dir, exist_ok=True)
            with open(f"{skill_dir}/__init__.py", "w") as f:
                f.write(prop.code)
            prop.status = "integrated"
            logger.info(f"✅ Compétence {skill.metadata.name} intégrée.")
            return True
        except Exception as e:
            logger.error(f"Erreur intégration skill : {e}")
            return False

    def reject_proposal(self, proposal_id: str) -> bool:
        prop = self.proposals.get(proposal_id)
        if not prop or prop.status != "pending":
            return False
        prop.status = "rejected"
        logger.info(f"❌ Proposition {proposal_id} rejetée.")
        return True