"""backend/api/decision_memory.py — GAP-A2: Decision Memory anti-ripetizione. Memorizza le decisioni dell'agente (fix proposti + esito) per evitare: - Re-proposizione di fix già rifiutati (blacklist) - Ri-analisi di bug già chiusi - Attivazione di strategie già fallite Storage: agent_memory (category='decision') — stesso schema, nessuna nuova tabella. API: POST /api/memory/decision — registra decisione GET /api/memory/decision — lista tutte GET /api/memory/decision/blacklist — solo fix rifiutati GET /api/memory/decision/check — controlla se fix è blacklistato DELETE /api/memory/decision/{id} — rimuove dalla blacklist """ from __future__ import annotations import asyncio import hashlib import json import time from typing import Optional from fastapi import APIRouter, HTTPException from pydantic import BaseModel import logging _logger = logging.getLogger("agente_ai") # S-BUGFIX logger = logging.getLogger("agente_ai.decision_memory") router = APIRouter(prefix="/api/memory/decision", tags=["decision-memory"]) # ─── In-memory store ───────────────────────────────────────────────────────── _decisions: dict[str, dict] = {} # signature → decision # ─── Models ────────────────────────────────────────────────────────────────── class DecisionIn(BaseModel): fix: str outcome: str # "accepted" | "rejected" | "partial" | "timeout" task_id: Optional[str] = None context: Optional[str] = None # goal/errore che ha generato il fix reason: Optional[str] = None # perché accettato/rifiutato # ─── Helpers ───────────────────────────────────────────────────────────────── def _sig(fix: str) -> str: """Firma normalizzata: sha256(fix trimmed lowercase[:200]).""" return hashlib.sha256(fix.strip().lower()[:200].encode()).hexdigest()[:16] # ─── Core API ──────────────────────────────────────────────────────────────── async def record_decision( fix: str, outcome: str, task_id: str = "", context: str = "", reason: str = "" ) -> str: """ Registra una decisione agente. Se esiste già una decisione con la stessa firma del fix, aggiorna outcome e incrementa il contatore. Fire-and-forget: non solleva mai eccezioni verso il chiamante. """ try: sig = _sig(fix) now_ms = int(time.time() * 1000) if sig in _decisions: d = _decisions[sig] d["outcome"] = outcome d["count"] += 1 d["last_seen"] = now_ms d["updated_at"] = now_ms if reason: d["reason"] = reason[:200] asyncio.create_task(_sb_save(d)).add_done_callback( lambda t: logger.debug("_sb_save/update exc: %s", t.exception()) if not t.cancelled() and t.exception() is not None else None ) return d["id"] did = f"dec_{sig[:8]}" decision = { "id": did, "signature": sig, "fix": fix[:500], "outcome": outcome, "task_id": (task_id or "")[:64], "context": (context or "")[:300], "reason": (reason or "")[:200], "count": 1, "first_seen": now_ms, "last_seen": now_ms, "created_at": now_ms, "updated_at": now_ms, } _decisions[sig] = decision asyncio.create_task(_sb_save(decision)).add_done_callback( lambda t: logger.debug("_sb_save/new exc: %s", t.exception()) if not t.cancelled() and t.exception() is not None else None ) logger.info("Decision %s: [%s] %s", did, outcome, fix[:60]) return did except Exception as _e: logger.debug("record_decision error: %s", _e) return "" def is_blacklisted(fix: str) -> tuple[bool, str]: """ Controlla se un fix è stato precedentemente rifiutato/fallito. Ritorna (True, reason) se blacklistato, (False, '') altrimenti. """ d = _decisions.get(_sig(fix)) if not d: return False, "" if d.get("outcome") in ("rejected", "timeout") and d.get("count", 0) >= 1: reason = d.get("reason") or f"Fix già {d['outcome']} {d['count']} volta/e" return True, reason return False, "" def get_blacklist() -> list[dict]: return sorted( [d for d in _decisions.values() if d.get("outcome") in ("rejected", "timeout")], key=lambda x: x.get("count", 0), reverse=True, ) # ─── Supabase persistence ───────────────────────────────────────────────────── async def _sb_save(decision: dict) -> None: try: # ─── Mirroring su Hugging Face Dataset (Zero-Cost Backup) ───────────── try: from .hf_storage import hf_fire_and_forget hf_fire_and_forget("decisions.jsonl", decision) except Exception as hf_exc: logger.debug("HF Mirroring silenced: %s", hf_exc) from .state import _sb if not _sb: return now = int(time.time() * 1000) await asyncio.to_thread( lambda: _sb.table("agent_memory").upsert( {"key": f"dec:{decision['id']}", "category": "decision", "value": json.dumps(decision, ensure_ascii=False)[:4000], "created_at": now, "updated_at": now}, on_conflict="key", ).execute() ) except Exception as exc: logger.debug("decision _sb_save: %s", exc) async def _sb_load() -> None: try: from .state import _sb if not _sb: return result = await asyncio.to_thread( lambda: _sb.table("agent_memory").select("key,value") .eq("category", "decision").execute() ) for row in (result.data or []): try: d = json.loads(row["value"]) if isinstance(d, dict) and "signature" in d: _decisions.setdefault(d["signature"], d) except Exception as _exc: _logger.debug("[decision_memory] silenced %s", type(_exc).__name__) # noqa: BLE001 logger.info("Decision memory: caricati %d record da Supabase", len(_decisions)) except Exception as exc: logger.warning("Decision memory Supabase load fallito: %s", exc) def start_decision_memory() -> None: """Avvia il registry. Chiamato in _on_startup() di main.py.""" try: loop = asyncio.get_event_loop() if loop.is_running(): asyncio.create_task(_sb_load()).add_done_callback( lambda t: logger.debug("_sb_load exc: %s", t.exception()) if not t.cancelled() and t.exception() is not None else None ) except Exception as exc: logger.warning("Decision memory start: %s", exc) # ─── REST Endpoints ────────────────────────────────────────────────────────── @router.post("", status_code=201) async def create_decision(body: DecisionIn) -> dict: did = await record_decision( fix=body.fix, outcome=body.outcome, task_id=body.task_id or "", context=body.context or "", reason=body.reason or "", ) return _decisions.get(_sig(body.fix), {"id": did}) @router.get("") async def list_all_decisions() -> list[dict]: return sorted(_decisions.values(), key=lambda x: x.get("updated_at", 0), reverse=True) @router.get("/blacklist") async def get_blacklist_endpoint() -> list[dict]: """Fix rifiutati — da consultare prima di ri-proporre soluzioni.""" return get_blacklist() @router.get("/check") async def check_fix(fix: str) -> dict: """Controlla se un fix è blacklistato. Usato dall'agente prima di proporlo.""" bl, reason = is_blacklisted(fix) return {"blacklisted": bl, "reason": reason, "fix": fix[:100]} @router.delete("/{decision_id}") async def remove_decision(decision_id: str) -> dict: """Rimuove decisione dalla blacklist (unblacklist manuale).""" sig = next((s for s, d in _decisions.items() if d.get("id") == decision_id), None) if not sig: raise HTTPException(404, f"Decisione {decision_id} non trovata") del _decisions[sig] try: from .state import _sb if _sb: await asyncio.to_thread( lambda: _sb.table("agent_memory").delete() .eq("key", f"dec:{decision_id}").execute() ) except Exception as _exc: _logger.debug("[decision_memory] silenced %s", type(_exc).__name__) # noqa: BLE001 return {"deleted": decision_id}