api/incident_registry.py

"""backend/api/incident_registry.py — GAP-A1: Incident Registry + ARCH-K2.6 provider health incidents.

Registro unico degli incidenti dell'agente con:
  - Causa radice identificata via euristica leggera (zero LLM)
  - Deduplicazione per firma errore: stessa firma → occurrences++
  - Status lifecycle: open → resolved | suppressed
  - Delta tracking: get_delta(since_ms) per viste incrementali (GAP-A7)

Storage (nessuna nuova tabella Supabase):
  - In-memory dict _incidents (fonte di verità nel processo)
  - agent_memory (category='incident') — sopravvive ai restart

API:
  GET  /api/incidents              — lista incidenti aperti
  GET  /api/incidents/delta        — delta da since_ms (GAP-A7)
  GET  /api/incidents/summary      — riepilogo per categoria
  POST /api/incidents              — crea manualmente
  PATCH /api/incidents/{id}/resolve — chiude incidente

ARCH-K2.6:
  log_provider_incident() — scorciatoia per incidenti provider health
"""
from __future__ import annotations

import asyncio
import hashlib
import json
import time
from typing import Any, Optional

from fastapi import APIRouter, Depends, HTTPException
from .auth_guard import require_role, AuthRole
from pydantic import BaseModel
import logging
_logger = logging.getLogger("agente_ai")  # S-BUGFIX

logger = logging.getLogger("agente_ai.incident_registry")
router = APIRouter(prefix="/api/incidents", tags=["incidents"], dependencies=[Depends(require_role(AuthRole.MACHINE))])  # GAP-1-fix: router-level auth

# ─── In-memory store ─────────────────────────────────────────────────────────
_incidents: dict[str, dict] = {}   # id → incident

# ─── Root-cause heuristics (zero LLM) ────────────────────────────────────────
_RC_RULES: list[tuple[list[str], str, str]] = [
    (["timeout", "timed out", "TimeoutError", "ReadTimeout"],
     "timeout",     "Operazione scaduta — aumenta timeout o ottimizza la query"),
    (["PermissionError", "403", "Forbidden", "access denied", "permission"],
     "permission",  "Permessi insufficienti — verifica token/chiavi API"),
    (["ConnectionError", "ConnectTimeout", "httpx.Connect", "socket", "network"],
     "network",     "Errore di rete — verifica connettività Railway → API esterna"),
    (["OpenAI", "Anthropic", "429", "rate limit", "quota", "RateLimitError"],
     "ai_quota",    "Quota AI esaurita — attendi o ruota provider"),
    (["Supabase", "postgres", "relation does not exist", "column", "database"],
     "database",    "Errore database — verifica schema Supabase e credenziali"),
    (["NameError", "AttributeError", "TypeError", "KeyError", "IndexError", "SyntaxError"],
     "code_bug",    "Bug nel codice — analizza traceback e correggi"),
    (["MemoryError", "OOM", "out of memory", "killed"],
     "memory",      "Memoria esaurita — riduci payload o ottimizza allocazione"),
    # ARCH-K2.6: Provider health categories
    (["hf space", "sleeping", "space is paused", "502", "503", "cold start", "warmup"],
     "hf_space",    "HF Space sospeso o in cold start — attendere il riavvio automatico o ridurre l'inactive timeout"),
    (["no provider", "no_provider", "circuit breaker", "cb_blocked", "execution fabric"],
     "provider_down", "Execution provider non disponibile — verificare URL e stato del servizio nel Fabric"),
    (["railway", "deploy failed", "railway timeout"],
     "railway",     "Railway backend non raggiungibile — verificare il deployment e i log Railway"),
]

def _classify(error: str) -> tuple[str, str]:
    e = error[:800].lower()
    for keywords, cat, cause in _RC_RULES:
        if any(k.lower() in e for k in keywords):
            return cat, cause
    return "internal", "Errore interno non classificato — analizza i log Railway"

def _sig(goal: str, error: str) -> str:
    cat, _ = _classify(error)
    return hashlib.sha256(f"{goal[:80]}:{cat}:{error[:120]}".encode()).hexdigest()[:16]

# ─── Core API (usata da agent.py e scheduler.py) ─────────────────────────────

async def log_incident(task_id: str, goal: str, error: str, source: str = "agent") -> str:
    """
    Registra un incidente. Se esiste già un incidente aperto con la stessa firma
    (stesso tipo di errore sullo stesso goal), incrementa occurrences e aggiorna
    last_seen invece di crearne uno nuovo. Ritorna l'incident_id.
    Fire-and-forget: non solleva mai eccezioni verso il chiamante.
    """
    try:
        sig = _sig(goal, error)
        now_ms = int(time.time() * 1000)
        cat, rc = _classify(error)
        existing = next((i for i in _incidents.values()
                         if i.get("signature") == sig and i.get("status") == "open"), None)
        if existing:
            existing["occurrences"] += 1
            existing["last_seen"]    = now_ms
            existing["updated_at"]   = now_ms
            asyncio.create_task(_sb_save(existing)).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
        )
            logger.info("Incident %s: occurrences=%d (%s)", existing["id"], existing["occurrences"], cat)
            return existing["id"]
        iid = f"inc_{sig[:8]}_{now_ms % 100000:05x}"
        incident: dict[str, Any] = {
            "id":          iid,
            "signature":   sig,
            "task_id":     task_id[:64],
            "goal":        goal[:300],
            "error":       error[:500],
            "category":    cat,
            "root_cause":  rc,
            "source":      source,
            "status":      "open",
            "occurrences": 1,
            "first_seen":  now_ms,
            "last_seen":   now_ms,
            "created_at":  now_ms,
            "updated_at":  now_ms,
            "resolved_at": None,
        }
        _incidents[iid] = incident
        asyncio.create_task(_sb_save(incident)).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("New incident %s: [%s] %s", iid, cat, goal[:60])
        return iid
    except Exception as _e:
        logger.debug("log_incident error: %s", _e)
        return ""


def resolve_incident(incident_id: str) -> bool:
    inc = _incidents.get(incident_id)
    if not inc or inc.get("status") != "open":
        return False
    now_ms = int(time.time() * 1000)
    inc.update({"status": "resolved", "resolved_at": now_ms, "updated_at": now_ms})
    asyncio.create_task(_sb_save(inc)).add_done_callback(_log_bg_inc).add_done_callback(
            lambda t: logger.debug("_sb_save/resolve exc: %s", t.exception())
            if not t.cancelled() and t.exception() is not None else None
        )
    return True


def get_open() -> list[dict]:
    return sorted(
        [i for i in _incidents.values() if i.get("status") == "open"],
        key=lambda x: x["last_seen"], reverse=True,
    )


def get_delta(since_ms: int) -> list[dict]:
    """Vista delta-only (GAP-A7): solo incidenti cambiati dopo since_ms."""
    return sorted(
        [i for i in _incidents.values() if i.get("updated_at", 0) > since_ms],
        key=lambda x: x["updated_at"], reverse=True,
    )


# ─── Supabase persistence (agent_memory, category='incident') ─────────────────

async def log_provider_incident(
    provider_id:   str,
    provider_name: str,
    health:        str,
    reason:        str,
    source:        str = "execution_fabric",
) -> str:
    """
    Scorciatoia per loggare incidenti di salute dei provider (ARCH-K2.6).
    Già deduplicati per firma — stessa coppia (provider, health) → occurrences++.
    """
    return await log_incident(
        task_id = f"fabric:{provider_id}",
        goal    = f"health monitor: {provider_name}",
        error   = f"Provider {provider_name} health={health}: {reason}",
        source  = source,
    )


async def _sb_save(incident: dict) -> None:
    try:
        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"inc:{incident['id']}", "category": "incident",
                 "value": json.dumps(incident, ensure_ascii=False)[:7000],
                 "created_at": now, "updated_at": now},
                on_conflict="key",
            ).execute()
        )
    except Exception as exc:
        logger.debug("incident _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", "incident").execute()
        )
        for row in (result.data or []):
            try:
                inc = json.loads(row["value"])
                if isinstance(inc, dict) and "id" in inc:
                    _incidents.setdefault(inc["id"], inc)
            except Exception as _exc:
                _logger.debug("[incident_registry] silenced %s", type(_exc).__name__)  # noqa: BLE001
        logger.info("Incident registry: caricati %d incidenti da Supabase", len(_incidents))
    except Exception as exc:
        logger.warning("Incident registry Supabase load fallito: %s", exc)


def start_incident_registry() -> None:
    """Avvia il registry (carica da Supabase). Chiamato in _on_startup() di main.py."""
    try:
        loop = asyncio.get_event_loop()
        if loop.is_running():
            # BUGFIX: eccezioni del task background erano perse silenziosamente
            def _log_bg_inc(t):
                if not t.cancelled() and t.exception():
                    _logger.warning("[incident_registry] bg task raised: %s", t.exception())
            asyncio.create_task(_sb_load()).add_done_callback(_log_bg_inc)
    except Exception as exc:
        logger.warning("Incident registry start: %s", exc)


# ─── REST Endpoints ──────────────────────────────────────────────────────────

class _IncidentCreate(BaseModel):
    task_id: str = "manual"
    goal:    str = ""
    error:   str = ""
    source:  str = "manual"

class _IncidentPatch(BaseModel):
    status: Optional[str] = None   # "resolved" | "suppressed"


@router.get("")
async def list_incidents() -> list[dict]:
    """Lista incidenti aperti, ordinati per last_seen desc."""
    return get_open()


@router.get("/delta")
async def incidents_delta(since_ms: int = 0) -> list[dict]:
    """Delta-only: incidenti cambiati dopo since_ms. Usare per polling incrementale."""
    return get_delta(since_ms)


@router.get("/summary")
async def incidents_summary() -> dict:
    """Riepilogo rapido: totale, per categoria, aperti vs risolti. Include provider health (ARCH-K2.6)."""
    by_cat: dict[str, int] = {}
    open_n = resolved_n = 0
    for inc in _incidents.values():
        cat = inc.get("category", "internal")
        by_cat[cat] = by_cat.get(cat, 0) + 1
        if inc.get("status") == "open":
            open_n += 1
        else:
            resolved_n += 1
    return {"total": len(_incidents), "open": open_n, "resolved": resolved_n, "by_category": by_cat}


@router.post("", status_code=201)
async def create_incident_endpoint(body: _IncidentCreate) -> dict:
    iid = await log_incident(body.task_id, body.goal, body.error, body.source)
    return _incidents.get(iid, {"id": iid})


@router.patch("/{incident_id}/resolve", status_code=200)
async def resolve_endpoint(incident_id: str) -> dict:
    ok = resolve_incident(incident_id)
    if not ok:
        raise HTTPException(404, f"Incidente {incident_id} non trovato o già risolto")
    return _incidents[incident_id]


@router.patch("/{incident_id}/suppress", status_code=200)
async def suppress_endpoint(incident_id: str) -> dict:
    """Sopprime un incidente (non mostrarlo più ma non perderlo)."""
    inc = _incidents.get(incident_id)
    if not inc:
        raise HTTPException(404, f"Incidente {incident_id} non trovato")
    now_ms = int(time.time() * 1000)
    inc.update({"status": "suppressed", "updated_at": now_ms})
    asyncio.create_task(_sb_save(inc))
    return inc