"""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