memory/reflection.py

"""
reflection.py — Reflection Memory
Impara dagli errori, pattern di fallimento, strategie vincenti.
"""
import json
from pathlib import Path
from datetime import datetime

import logging
_logger = logging.getLogger("memory.reflection")

# D1: usa /data/ se disponibile (HF Spaces volume persistente) — evita perdita a restart
import os as _os_ref
_REFLECT_DATA_DIR = _os_ref.getenv('CHROMA_DATA_DIR') or ('/data' if Path('/data').exists() else '.')
REFLECTION_PATH   = Path(_REFLECT_DATA_DIR) / 'reflection_memory.json'

class ReflectionMemory:
    def __init__(self):
        self._data: dict = {"failures": [], "successes": [], "patterns": {}}
        self._load()

    def _load(self):
        if REFLECTION_PATH.exists():
            try:
                self._data = json.loads(REFLECTION_PATH.read_text())
            except Exception as _exc:
                _logger.debug("[reflection] silenced %s", type(_exc).__name__)  # noqa: BLE001

    def _save(self):
        REFLECTION_PATH.write_text(json.dumps(self._data, indent=2, ensure_ascii=False))

    def record_failure(self, task: str, error: str, attempted_strategy: str = ""):
        entry = {
            "ts": datetime.now().isoformat(),
            # S600: task 300→500 — task description può essere più lunga
            "task": task[:500],
            "error": error[:500],
            "strategy": attempted_strategy,
        }
        self._data["failures"].append(entry)
        # Track pattern frequency
        key = error[:120].lower()  # S582: 80→120 (chiave pattern freq — conservativo)
        self._data["patterns"][key] = self._data["patterns"].get(key, 0) + 1
        # Cap at 200 entries
        self._data["failures"] = self._data["failures"][-200:]
        self._save()

    def record_success(self, task: str, strategy: str):
        entry = {
            "ts": datetime.now().isoformat(),
            # S600: task 300→500, strategy 300→500 — più contesto per replay futuro
            "task": task[:500],
            "strategy": strategy[:500],
        }
        self._data["successes"].append(entry)
        self._data["successes"] = self._data["successes"][-100:]
        self._save()

    def get_relevant_lessons(self, task: str, n: int = 3) -> list[dict]:
        """Find past failures/successes relevant to current task."""
        task_lower = task.lower()
        words = set(task_lower.split())
        def score(entry):
            text = (entry.get("task", "") + " " + entry.get("strategy", "") + entry.get("error", "")).lower()
            return sum(1 for w in words if w in text and len(w) > 3)

        failures = sorted(self._data["failures"], key=score, reverse=True)[:n]
        successes = sorted(self._data["successes"], key=score, reverse=True)[:n]
        lessons = []
        for f in failures[:2]:
            if score(f) > 0:
                lessons.append({"type": "failure", "task": f["task"], "avoid": f["error"]})
        for s in successes[:1]:
            if score(s) > 0:
                lessons.append({"type": "success", "task": s["task"], "strategy": s["strategy"]})
        return lessons

    def get_top_patterns(self, n: int = 5) -> list[dict]:
        return sorted(
            [{"pattern": k, "count": v} for k, v in self._data["patterns"].items()],
            key=lambda x: x["count"], reverse=True
        )[:n]

    def stats(self) -> dict:
        return {
            "failures": len(self._data["failures"]),
            "successes": len(self._data["successes"]),
            "patterns": len(self._data["patterns"]),
        }