""" error_classifier.py — S404: Classificazione errori nel repair loop Documento strategico: "Non basta 'vedere un errore'. Devi distinguere almeno: selettore sbagliato, pagina diversa dal previsto, login fallito, errore runtime, rete lenta, frame non caricato, stato UI incoerente." Implementa la classificazione per consentire repair MIRATI invece di analisi generiche. Integrazione: _reflective_debug → classify_error(goal, errors) → ErrorResult → inject repair_strategy nel context prima del retry Pipeline classificazione: 1. Pattern matching regex (deterministico, zero latenza) 2. Fallback: UNKNOWN → generic strategic recovery Non usa LLM per classificare — è sincrono e costa zero latency. """ import re from dataclasses import dataclass from enum import Enum class ErrorCategory(str, Enum): SELECTOR = "selector" # CSS/DOM selector non trovato NAVIGATION = "navigation" # URL sbagliato, pagina non trovata, redirect AUTH = "auth" # Login fallito, token scaduto, 401/403 RUNTIME = "runtime" # TypeError, ReferenceError, is not defined NETWORK = "network" # Timeout, ECONNREFUSED, offline, 503 FRAME = "frame" # iframe non caricato, frame non trovato SYNTAX = "syntax" # SyntaxError, IndentationError, parse error LOGIC = "logic" # AssertionError, valore atteso ≠ ottenuto LIMIT = "limit" # Rate limit, quota, 429, OOM DB_ERROR = "db_error" # Sprint 3b: IntegrityError, FK, connection pool, deadlock UNKNOWN = "unknown" # Non classificato # ── Strategie di repair per categoria ───────────────────────────────────────── _REPAIR_STRATEGIES: dict[ErrorCategory, str] = { ErrorCategory.SELECTOR: ( "SELETTORE DOM: usa ID espliciti (`#id`), aria-label (`[aria-label='...']`) " "o ruoli ARIA (`getByRole('button', {name:'...'})`) — evita classi dinamiche. " "Prima leggi il DOM attuale per trovare il selettore corretto." ), ErrorCategory.NAVIGATION: ( "NAVIGAZIONE: verifica l'URL esatto prima di navigare. " "Attendi `domcontentloaded` o un elemento specifico prima di operare sulla pagina. " "Gestisci redirect e pagine di errore (404/500) esplicitamente." ), ErrorCategory.AUTH: ( "AUTENTICAZIONE: gestisci il login PRIMA di qualsiasi azione protetta. " "Verifica il token, gestisci la scadenza della sessione e i redirect post-login. " "Non assumere che la sessione sia già attiva." ), ErrorCategory.RUNTIME: ( "ERRORE RUNTIME: controlla undefined/null con optional chaining (`?.`) " "o guardie esplicite (`if (x == null) return`). " "Verifica i tipi degli argomenti prima di usarli. " "Aggiungi try/catch attorno alle operazioni rischiose." ), ErrorCategory.NETWORK: ( "ERRORE RETE: aggiungi retry con exponential backoff (max 3 tentativi). " "Gestisci offline/timeout con fallback utili. " "Non bloccare la UI durante le chiamate — usa stato di loading." ), ErrorCategory.FRAME: ( "IFRAME: aspetta che l'iframe sia caricato prima di accedere al suo DOM " "(usa `waitForLoadState` o `frameLocator`). " "Verifica che il frame esista con `page.frames()` prima di operare." ), ErrorCategory.SYNTAX: ( "SINTASSI: correggi l'errore di sintassi PRIMA di qualsiasi altra cosa. " "Verifica l'indentazione (Python), le parentesi bilanciate e i punti e virgola (JS). " "Usa un linter per identificare tutti gli errori nella stessa sessione." ), ErrorCategory.LOGIC: ( "LOGICA: rivedi i casi edge e le assunzioni. " "Aggiungi asserzioni esplicite sui valori attesi. " "Traccia il flusso di dati passo per passo per trovare dove diverge dal previsto." ), ErrorCategory.LIMIT: ( "RATE LIMIT / QUOTA: implementa retry con backoff esponenziale e jitter. " "Riduci la frequenza delle chiamate. " "Considera caching dei risultati per evitare chiamate ridondanti." ), ErrorCategory.DB_ERROR: ( "ERRORE DATABASE: controlla vincoli di integrità (FK, UNIQUE, NOT NULL). " "Per IntegrityError: verifica che i dati rispettino i vincoli prima dell'insert. " "Per connection pool exhausted: chiudi le connessioni non usate, riduci max_connections. " "Per deadlock: usa retry con backoff, considera SKIP LOCKED per job queue. " "Per 'relation does not exist': esegui le migrazioni pendenti prima dell'avvio." ), ErrorCategory.UNKNOWN: ( "Analizza il contesto completo dell'errore. " "Prova un approccio COMPLETAMENTE diverso da quello usato finora." ), } # ── Pattern regex per classificazione ───────────────────────────────────────── # Ordinati per priorità (il primo match vince) _PATTERNS: list[tuple[ErrorCategory, re.Pattern]] = [ (ErrorCategory.SYNTAX, re.compile( r"SyntaxError|IndentationError|ParseError|parse error|" r"unexpected token|unexpected EOF|invalid syntax|" r"unterminated string|missing \)", re.IGNORECASE, )), (ErrorCategory.SELECTOR, re.compile( r"selector|querySelector|getElementById|no element|" r"element not found|locator|aria|getByRole|" r"TimeoutError.*waiting for|strict mode violation", re.IGNORECASE, )), (ErrorCategory.AUTH, re.compile( r"401|403|Unauthorized|Forbidden|Authentication|" r"token.*expired|session.*expired|login.*failed|" r"invalid.*credentials|permission denied", re.IGNORECASE, )), (ErrorCategory.NETWORK, re.compile( r"ECONNREFUSED|ENOTFOUND|ETIMEDOUT|timeout|" r"network error|fetch.*failed|connection refused|" r"503|502|504|unreachable|offline", re.IGNORECASE, )), (ErrorCategory.LIMIT, re.compile( r"429|rate.?limit|quota.*exceeded|too many requests|" r"OOM|out of memory|memory.*error", re.IGNORECASE, )), (ErrorCategory.FRAME, re.compile( r"frame|iframe|frameLocator|frame.*not found|" r"cross.?origin|sandboxed.*frame", re.IGNORECASE, )), (ErrorCategory.NAVIGATION, re.compile( r"404|page not found|navigation.*failed|goto.*timeout|" r"ERR_NAME_NOT_RESOLVED|net::ERR|redirect.*loop|" r"wrong.*page|URL.*invalid", re.IGNORECASE, )), (ErrorCategory.RUNTIME, re.compile( r"TypeError|ReferenceError|is not defined|" r"Cannot read|Cannot set|undefined is not|" r"null.*property|property.*null|" r"is not a function|is not iterable|" r"UnhandledPromiseRejection", re.IGNORECASE, )), (ErrorCategory.LOGIC, re.compile( r"AssertionError|assertion.*failed|expected.*got|" r"mismatch|wrong.*value|incorrect.*result|" r"test.*failed|expected.*received", re.IGNORECASE, )), # Sprint 3b: DB_ERROR — IntegrityError, FK, connection pool, deadlock (ErrorCategory.DB_ERROR, re.compile( r"IntegrityError|ForeignKey|ForeignKeyViolation|" r"connection pool|deadlock|" r"relation.*not.*exist|table.*not.*exist|" r"duplicate key|violates.*constraint|" r"UniqueViolation|CheckViolation|NotNullViolation|" r"OperationalError.*database|psycopg|asyncpg|" r"FATAL.*database|could not connect.*database", re.IGNORECASE, )), ] @dataclass class ErrorResult: category: ErrorCategory repair_strategy: str confidence: float # 0.0-1.0 — quanto siamo sicuri della classificazione matched_pattern: str # substring che ha fatto match (debug) def classify_error(errors: list[str]) -> ErrorResult: """ Classifica una lista di messaggi di errore nella categoria più probabile. Algoritmo: 1. Concatena gli errori (max 1500 chars) 2. Prova i pattern in ordine di priorità (il primo match vince) 3. Se nessun match → UNKNOWN con confidence 0.0 Non usa LLM — è sincrono e a zero latency. Chiamato da _reflective_debug prima di invocare ARCHITECT. """ combined = " | ".join(str(e)[:500] for e in errors[-4:])[:1500] # S604: 400→500 per error string completa for category, pattern in _PATTERNS: m = pattern.search(combined) if m: matched = m.group(0)[:60] strategy = _REPAIR_STRATEGIES[category] return ErrorResult( category = category, repair_strategy = strategy, confidence = 0.85, matched_pattern = matched, ) return ErrorResult( category = ErrorCategory.UNKNOWN, repair_strategy = _REPAIR_STRATEGIES[ErrorCategory.UNKNOWN], confidence = 0.0, matched_pattern = "", ) def format_for_context(result: ErrorResult) -> str: """ Produce la stringa da iniettare nel context del loop prima del retry. Formato: [ERRORE CLASSIFICATO: ] + strategia. """ label = f"ERRORE CLASSIFICATO: {result.category.value.upper()}" if result.matched_pattern: label += f" (match: \"{result.matched_pattern}\")" return f"\n\n[{label}]\n{result.repair_strategy}"