""" requirement_engine.py — Sprint 2: Decomposizione semantica del goal in requisiti strutturati Input: goal string (es. "crea un CRM con login e dashboard") Output: lista requisiti [{id, feature, description, acceptance_criteria}] Strategia: 1. Pattern matching regex deterministico su goal common (zero LLM, zero latenza) 2. Fallback LLM (Groq fast, 1 call, 3s timeout) per goal non coperti dai pattern 3. Cache locale dict-session per goal già decomposed — niente LLM repeat calls Invarianti rispettate: - PR2: non tocca providerChain.ts (questo è solo backend) - B1: nessun corpo duplicato - Additive-only: fallback a lista vuota su qualsiasi errore — nessuna regressione """ from __future__ import annotations import re import json import asyncio import hashlib from dataclasses import dataclass, field from typing import Any # ── Cache session-level ──────────────────────────────────────────────────────── _REQUIREMENT_CACHE: dict[str, list[dict]] = {} _CACHE_MAX = 50 def _cache_key(goal: str) -> str: return hashlib.md5(goal.strip()[:300].lower().encode()).hexdigest() # ── Libreria pattern predefiniti ─────────────────────────────────────────────── # Ogni voce: (regex_pattern, feature_id, feature_name, description) _FEATURE_PATTERNS: list[tuple[re.Pattern, str, str, str]] = [ ( re.compile(r'\b(auth|login|logout|registr|signup|sign.?up|accesso|autenticaz)\b', re.I), "auth", "Autenticazione", "Login, logout, registrazione utente e gestione sessione", ), ( re.compile(r'\b(crud|create|read|update|delete|gestione|manage|list|detail|edit|modifica|elimina)\b', re.I), "crud", "Operazioni CRUD", "Creazione, lettura, modifica e cancellazione di entità dati", ), ( re.compile(r'\b(dashboard|overview|riepilogo|summary|stats|statistiche|kpi|metriche|analytics)\b', re.I), "dashboard", "Dashboard", "Vista aggregata con metriche, statistiche e KPI principali", ), ( re.compile(r'\b(api|rest|endpoint|route|backend|server|fastapi|flask|express|django)\b', re.I), "api_rest", "API REST", "Endpoint REST con validazione input e gestione errori", ), ( re.compile(r'\b(form|validaz|validation|input|campi|fields|submit|invio)\b', re.I), "form_validation", "Form e Validazione", "Form con validazione lato client e server", ), ( re.compile(r'\b(upload|file|immagine|image|media|attachment|allegato)\b', re.I), "file_upload", "Upload File", "Caricamento e gestione file/media", ), ( re.compile(r'\b(search|cerca|ricerca|filtro|filter|sort|ordinamento)\b', re.I), "search", "Ricerca e Filtri", "Ricerca full-text e filtraggio risultati", ), ( re.compile(r'\b(pagament|payment|stripe|checkout|cart|carrello|acquisto|order|ordine)\b', re.I), "payments", "Pagamenti", "Flusso di checkout e gestione pagamenti", ), ( re.compile(r'\b(notif|notification|alert|email|sms|push|avviso|messaggio)\b', re.I), "notifications", "Notifiche", "Sistema di notifiche e comunicazioni", ), ( re.compile(r'\b(setting|impostaz|profilo|profile|account|preferenze|config)\b', re.I), "settings", "Impostazioni", "Gestione profilo utente e impostazioni applicazione", ), ( re.compile(r'\b(database|db|schema|migration|tabella|table|model|entità|entity)\b', re.I), "database", "Database", "Schema dati, migrazioni e modelli", ), ( re.compile(r'\b(deploy|ci.?cd|docker|container|hosting|production|prod)\b', re.I), "deploy", "Deploy", "Pipeline di build e deploy in produzione", ), ( re.compile(r'\b(analizza|analyze|analisi|analytic|analisi comparativa|analyse)\b', re.I), "analysis", "Analisi", "Analisi dettagliata con argomentazioni, dati e conclusioni strutturate", ), ( re.compile(r'\b(confronta|compare|versus|vs\.?|differenz[ae]|differences?|migliore tra|meglio tra)\b', re.I), "comparison", "Confronto", "Confronto strutturato con dimensioni esplicite e raccomandazione finale", ), ( re.compile(r"\b(spiega|explain|descrivi|describe|cos[\'è ]{1,3}è|what is|how does|come funziona)\b", re.I), "explanation", "Spiegazione", "Spiegazione chiara con definizione, esempi concreti e contesto d'uso", ), ( re.compile(r'\b(riassumi|summarize|sommario|sintesi|riepilog[ao]|riassunto)\b', re.I), "summarization", "Sintesi", "Sintesi strutturata che mantiene i punti chiave senza perdita di informazioni critiche", ), ( re.compile(r'\b(raccomand[ai]|recommend|consigli[ao]|suggerisci|suggest|best practice|cosa sceglier|quale sceglier)\b', re.I), "recommendation", "Raccomandazione", "Raccomandazione motivata con almeno 3 criteri di valutazione e conclusione esplicita", ), ] # Criteri accettazione standard per ogni feature (importati da AcceptanceCriteria) from agents.acceptance_criteria import ACCEPTANCE_CRITERIA @dataclass class Requirement: id: str feature: str description: str acceptance_criteria: list[str] = field(default_factory=list) source: str = "pattern" # "pattern" | "llm" class RequirementEngine: """ Decompone un goal in lista di requisiti strutturati con criteri di accettazione. Chiamato da _run_fallback() su task complessi (_tok_budget >= 4096). Zero LLM per goal coperti dai pattern — fallback LLM solo per goal esotici. """ def __init__(self, llm: Any = None) -> None: self.llm = llm def decompose_sync(self, goal: str) -> list[Requirement]: """ Decomposizione sincrona via pattern matching. Usato quando non si è in un contesto async. """ key = _cache_key(goal) if key in _REQUIREMENT_CACHE: cached = _REQUIREMENT_CACHE[key] return [Requirement(**r) for r in cached] reqs = self._match_patterns(goal) self._store_cache(key, reqs) return reqs async def decompose(self, goal: str) -> list[Requirement]: """ Decomposizione async: pattern first, poi LLM fallback se lista vuota e goal complesso. """ key = _cache_key(goal) if key in _REQUIREMENT_CACHE: cached = _REQUIREMENT_CACHE[key] return [Requirement(**r) for r in cached] reqs = self._match_patterns(goal) # Fallback LLM solo se: nessun pattern matched + goal complesso (>30 chars) if not reqs and self.llm and len(goal.strip()) > 30: try: reqs = await asyncio.wait_for(self._decompose_llm(goal), timeout=3.0) except Exception: reqs = [] self._store_cache(key, reqs) return reqs def _match_patterns(self, goal: str) -> list[Requirement]: """Pattern matching deterministico — zero latenza.""" seen: set[str] = set() result: list[Requirement] = [] for pattern, feat_id, feat_name, description in _FEATURE_PATTERNS: if feat_id in seen: continue if pattern.search(goal): criteria = ACCEPTANCE_CRITERIA.get(feat_id, []) result.append(Requirement( id=feat_id, feature=feat_name, description=description, acceptance_criteria=criteria, source="pattern", )) seen.add(feat_id) return result async def _decompose_llm(self, goal: str) -> list[Requirement]: """ LLM fallback per goal non coperti dai pattern. 1 call, max 3s, output JSON strutturato. """ system = ( "Sei un analista di requisiti software. Dato un goal, estrai i requisiti " "in formato JSON. Rispondi SOLO con JSON valido, nessun testo aggiuntivo.\n" 'Formato: [{"id":"snake_case","feature":"Nome","description":"desc breve"}]' "\nMax 5 requisiti. Solo requisiti CONCRETI e VERIFICABILI." ) msgs = [ {"role": "system", "content": system}, # S596: goal 300→500 — goal complessi superano 300 chars {"role": "user", "content": f"Goal: {goal[:500]}"}, ] try: # S586: 300→512 — JSON array di requisiti con 3-5 items supera 300 tok raw = await self.llm.chat(msgs, temperature=0.0, max_tokens=512) if not raw or raw.startswith("[LLM"): return [] m = re.search(r'\[[\s\S]+\]', raw) if not m: return [] items = json.loads(m.group()) result = [] for item in items[:5]: if not isinstance(item, dict): continue feat_id = str(item.get("id", "unknown")) criteria = ACCEPTANCE_CRITERIA.get(feat_id, []) result.append(Requirement( id=feat_id, feature=str(item.get("feature", feat_id)), description=str(item.get("description", ""))[:400], # S576: 200→400 acceptance_criteria=criteria, source="llm", )) return result except Exception: return [] @staticmethod def _store_cache(key: str, reqs: list[Requirement]) -> None: """Mantieni cache sotto _CACHE_MAX entries.""" if len(_REQUIREMENT_CACHE) >= _CACHE_MAX: oldest = next(iter(_REQUIREMENT_CACHE)) _REQUIREMENT_CACHE.pop(oldest, None) _REQUIREMENT_CACHE[key] = [ { "id": r.id, "feature": r.feature, "description": r.description, "acceptance_criteria": r.acceptance_criteria, "source": r.source, } for r in reqs ] @staticmethod def format_for_context(reqs: list[Requirement]) -> str: """Formatta i requisiti per l'injection nel system prompt.""" if not reqs: return "" lines = ["[REQUISITI DECOMPOSED]"] for r in reqs: lines.append(f"• {r.feature}: {r.description}") lines.append( "\nIl GoalVerifier verificherà CIASCUN requisito separatamente. " "Assicurati che la risposta copra tutti i punti elencati." ) return "\n".join(lines)