agents/reflection_sidecar.py

"""
reflection_sidecar.py — Reflection Sidecar (Double-Token Innovation)

Analizza i log di errore di ogni tool call durante la sessione e aggiorna
session_rules.md in tempo reale. Questo file viene iniettato nel system prompt
dell'agente principale per correggere il comportamento on-the-fly.

Architettura:
  Token A (agente principale) → esegue tool, chiama log_error()
  Token B (sidecar critic)    → analizza pattern, scrive regole
"""

from __future__ import annotations

import asyncio
import json
import logging
import os
import re
import time
from collections import defaultdict, deque
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any

_logger = logging.getLogger("agente_ai.reflection_sidecar")

# ── Config ────────────────────────────────────────────────────────────────────

_RULES_FILE = Path(os.getenv("SIDECAR_RULES_FILE", "/data/session_rules.md"))
_MAX_ERRORS_BEFORE_REFLECT = int(os.getenv("SIDECAR_REFLECT_THRESHOLD", "2"))
_RULE_TTL_S = int(os.getenv("SIDECAR_RULE_TTL_S", "3600"))  # 1h

# Token B per NVIDIA NIM (Reflection Critic — modello leggero, bassa latenza)
_NVIDIA_API   = "https://integrate.api.nvidia.com/v1"
_CRITIC_MODEL = os.getenv("NVIDIA_B_MODEL", "meta/llama-3.3-70b-instruct")
_NVIDIA_KEY_B = os.getenv("NVIDIA_API_KEY_B", "")


# ── Data model ────────────────────────────────────────────────────────────────

@dataclass
class ErrorEvent:
    tool: str
    error: str
    context: str
    ts: float = field(default_factory=time.monotonic)


@dataclass
class SessionRule:
    pattern: str          # cosa ha causato l'errore (regex / descrizione)
    rule: str             # istruzione correttiva per l'agente
    tool: str
    created_at: float = field(default_factory=time.time)
    hit_count: int = 0


# ── Sidecar core ─────────────────────────────────────────────────────────────

class ReflectionSidecar:
    """
    Singleton per sessione. Riceve errori, li analizza con Token B (NVIDIA),
    aggiorna session_rules.md che viene iniettato nel prompt principale.
    """

    def __init__(self) -> None:
        self._errors: list[ErrorEvent] = []
        self._rules: list[SessionRule] = []
        self._tool_error_counts: dict[str, int] = defaultdict(int)
        self._lock = asyncio.Lock()
        self._reflect_task: asyncio.Task | None = None

    async def log_error(
        self,
        tool: str,
        error: str,
        context: str = "",
    ) -> None:
        """Registra un errore. Se lo stesso tool fallisce >= threshold, avvia reflection."""
        async with self._lock:
            evt = ErrorEvent(tool=tool, error=error[:500], context=context[:300])
            self._errors.append(evt)
            self._tool_error_counts[tool] += 1
            count = self._tool_error_counts[tool]

        if count >= _MAX_ERRORS_BEFORE_REFLECT:
            # Avvia reflection in background (non blocca l'agente principale)
            if self._reflect_task is None or self._reflect_task.done():
                # BUGFIX: eccezioni di _reflect_and_update erano perse silenziosamente
                def _log_ref_exc(t):
                    if not t.cancelled() and t.exception():
                        _logger.warning("[reflection_sidecar] reflect task raised: %s", t.exception())
                self._reflect_task = asyncio.create_task(
                    self._reflect_and_update(tool, error, context)
                )
                self._reflect_task.add_done_callback(_log_ref_exc)

    async def _reflect_and_update(
        self, tool: str, last_error: str, context: str
    ) -> None:
        """Token B: analizza gli errori e genera una regola correttiva."""
        if not _NVIDIA_KEY_B:
            _logger.warning("reflection_sidecar: NVIDIA_API_KEY_B non configurato — skip")
            return

        # Aggrega tutti gli errori del tool
        relevant = [e for e in self._errors if e.tool == tool][-5:]
        error_summary = "\n".join(f"- [{e.tool}] {e.error}" for e in relevant)

        prompt = f"""Sei un critico di qualità per un agente AI. Analizza questi errori ripetuti:

TOOL: {tool}
ERRORI:
{error_summary}

CONTESTO ULTIMO ERRORE: {context}

Scrivi UNA regola correttiva concisa (max 2 righe) che l'agente deve seguire per evitare
di ripetere questo errore. Formato: "REGOLA [{tool}]: <istruzione diretta all'agente>"
Rispondi solo con la regola, nessun altro testo."""

        try:
            import urllib.request
            payload = json.dumps({
                "model": _CRITIC_MODEL,
                "messages": [{"role": "user", "content": prompt}],
                "max_tokens": 120,
                "temperature": 0.1,
            }).encode()
            req = urllib.request.Request(
                f"{_NVIDIA_API}/chat/completions",
                data=payload,
                headers={
                    "Authorization": f"Bearer {_NVIDIA_KEY_B}",
                    "Content-Type": "application/json",
                },
                method="POST",
            )
            with urllib.request.urlopen(req, timeout=15) as resp:
                data = json.loads(resp.read())
            rule_text = data["choices"][0]["message"]["content"].strip()

            new_rule = SessionRule(
                pattern=last_error[:100],
                rule=rule_text,
                tool=tool,
            )
            async with self._lock:
                # Dedup: rimuovi regole vecchie per lo stesso tool
                self._rules = [r for r in self._rules if r.tool != tool]
                self._rules.append(new_rule)
                self._tool_error_counts[tool] = 0  # reset counter

            await self._write_rules_file()
            _logger.info(f"reflection_sidecar: nuova regola generata per {tool}")

        except Exception as exc:
            _logger.warning(f"reflection_sidecar: reflection fallita — {exc}")

    async def _write_rules_file(self) -> None:
        """Scrive session_rules.md — viene iniettato nel system prompt principale."""
        now = time.time()
        active = [r for r in self._rules if (now - r.created_at) < _RULE_TTL_S]
        if not active:
            return
        lines = ["# Session Rules (auto-generate dal Reflection Sidecar)\n"]
        lines += [f"- {r.rule}" for r in active]
        lines.append(f"\n_Aggiornato: {time.strftime('%H:%M:%S')}_")
        try:
            _RULES_FILE.parent.mkdir(parents=True, exist_ok=True)
            _RULES_FILE.write_text("\n".join(lines), encoding="utf-8")
        except Exception as exc:
            _logger.warning(f"reflection_sidecar: scrittura rules file fallita — {exc}")

    def get_rules_for_prompt(self) -> str:
        """Legge session_rules.md per l'iniezione nel system prompt."""
        try:
            if _RULES_FILE.exists():
                content = _RULES_FILE.read_text(encoding="utf-8").strip()
                if content and len(content) > 30:
                    return f"\n\n---\n{content}\n---"
        except Exception:
            pass
        return ""

    def reset(self) -> None:
        """Reset a inizio nuova sessione."""
        self._errors.clear()
        self._rules.clear()
        self._tool_error_counts.clear()
        try:
            _RULES_FILE.unlink(missing_ok=True)
        except Exception:
            pass


# ── Singleton ─────────────────────────────────────────────────────────────────
_sidecar: ReflectionSidecar | None = None


def get_sidecar() -> ReflectionSidecar:
    global _sidecar
    if _sidecar is None:
        _sidecar = ReflectionSidecar()
    return _sidecar


async def log_tool_error(tool: str, error: str, context: str = "") -> None:
    """Shortcut globale — chiamare dopo ogni tool call fallita."""
    await get_sidecar().log_error(tool, error, context)


def get_session_rules() -> str:
    """Shortcut globale — iniettare nel system prompt principale."""
    return get_sidecar().get_rules_for_prompt()