memory/evolutionary.py

"""
backend/memory/evolutionary.py — Evolutionary Memory (S960)
Distilla preferenze utente, stili di codice e regole operative dalle sessioni.
Alimenta il layer 'Reflection' con conoscenze di alto livello (Long-term Evolution).
"""
import json
import logging
import os
from pathlib import Path
from datetime import datetime
from typing import List, Dict, Any

_logger = logging.getLogger("memory.evolutionary")

# Directory per i dati evolutivi (Sync con reflection.py)
_DATA_DIR = os.getenv('CHROMA_DATA_DIR') or ('/data' if Path('/data').exists() else '.')
EVO_PATH  = Path(_DATA_DIR) / 'evolutionary_rules.json'

class EvolutionaryMemory:
    def __init__(self, ai_client=None):
        self.ai_client = ai_client
        self.rules: Dict[str, Any] = {
            "user_preferences": {},  # es. "language": "python", "style": "functional"
            "operational_rules": [], # es. "Usa sempre pnpm invece di npm"
            "domain_knowledge": {},  # es. "path/to/project": "description"
            "last_updated": None
        }
        self._load()

    def _load(self):
        if EVO_PATH.exists():
            try:
                self.rules = json.loads(EVO_PATH.read_text())
            except Exception as e:
                _logger.error(f"[S960] Load error: {e}")

    def _save(self):
        try:
            EVO_PATH.write_text(json.dumps(self.rules, indent=2, ensure_ascii=False))
        except Exception as e:
            _logger.error(f"[S960] Save error: {e}")

    async def distill_and_evolve(self, session_summary: Dict[str, Any]):
        """
        Prende un sommario distillato (dal MemoryDistiller) e aggiorna le regole evolutive.
        """
        # 1. Estrazione euristica (in attesa di LLM integration)
        # Se il sommario contiene fatti chiave, li integriamo
        facts = session_summary.get("facts", [])
        for fact in facts:
            if ":" in fact:
                k, v = fact.split(":", 1)
                self.rules["domain_knowledge"][k.strip()] = v.strip()

        # 2. Rilevamento preferenze (es. linguaggi usati con successo)
        lessons = session_summary.get("lessons", [])
        for lesson in lessons:
            if lesson.get("type") == "success":
                # Esempio: "Usato FastAPI con successo" -> preferenza per FastAPI
                pass

        self.rules["last_updated"] = datetime.now().isoformat()
        self._save()
        _logger.info("[S960] Memoria evolutiva aggiornata.")

    def get_evolutionary_context(self) -> str:
        """
        Ritorna una stringa formattata da iniettare nel System Prompt.
        """
        if not self.rules["user_preferences"] and not self.rules["operational_rules"] and not self.rules["domain_knowledge"]:
            return ""

        context = "\n[MEMORIA EVOLUTIVA - REGOLE APPRESE]\n"
        
        if self.rules["user_preferences"]:
            context += "Preferenze Utente:\n"
            for k, v in self.rules["user_preferences"].items():
                context += f"- {k}: {v}\n"
                
        if self.rules["operational_rules"]:
            context += "Regole Operative:\n"
            for rule in self.rules["operational_rules"]:
                context += f"- {rule}\n"
                
        if self.rules["domain_knowledge"]:
            context += "Conoscenza Dominio:\n"
            for k, v in self.rules["domain_knowledge"].items():
                context += f"- {k}: {v}\n"
                
        return context

# Singleton
evo_memory = EvolutionaryMemory()