""" context_manager.py — Intelligent Context Management (S364) Implementa il "Project Skeleton" approach: - Skeleton aggiornato (nomi file + firme funzioni) sempre disponibile - Full content solo per file attivamente modificati - File "cold" riepilogatati con CONTEXT role (Groq-8b-instant) Risolve il "lost in the middle" problem su sessioni lunghe. Design: stateless per request, tutto I/O fire-and-forget, mai blocca il loop S752-A: aggiunta rank_files_by_relevance() — top-K selezione per rilevanza goal. FIX-SKEL-RAG: symbol matching + fuzzy prefix + zero-score filter + path weight 2.0. FIX-SYN-EXPAND: synonym expansion IT/EN per copertura semantica senza embeddings. """ from __future__ import annotations import asyncio import hashlib import re from typing import Any _FUNC_RE = re.compile( r'^(?:export\s+)?(?:async\s+)?(?:function\s+(\w+)|const\s+(\w+)\s*=\s*(?:async\s*)?\()', re.MULTILINE) _CLASS_RE = re.compile(r'^(?:export\s+)?class\s+(\w+)', re.MULTILINE) _PY_DEF_RE = re.compile(r'^(?: )?(?:async\s+)?def\s+(\w+)\s*\(', re.MULTILINE) _PY_CLS_RE = re.compile(r'^class\s+(\w+)', re.MULTILINE) _SUMMARY_CACHE: dict[str, str] = {} _MAX_SUMMARY_CACHE = 200 # ── S752-A: stopword set per rank_files_by_relevance ────────────────────────── _RANK_STOP_IT = { 'il','lo','la','i','gli','le','di','del','della','dei','delle', 'in','un','una','uno','che','con','per','non','da','si','su','al', 'ci','e','a','tra','fra','ma','o','se','ne','ad','ho','ha','è', } _RANK_STOP_EN = { 'the','a','an','in','on','at','to','for','of','and','or','is', 'are','was','be','this','that','it','with','as','by','from','about', 'can','will','have','has','had','do','does','did','not','but','if', } _RANK_STOP = _RANK_STOP_IT | _RANK_STOP_EN # Entry-point / config files ottengono un piccolo boost di rilevanza _RANK_ENTRY_STEMS = {'main', 'index', 'app', '__init__', 'config', 'settings', 'routes'} # ── FIX-SKEL-RAG: helper per fuzzy prefix matching ──────────────────────────── _CAMEL_SPLIT_RE = re.compile(r'([a-z])([A-Z])') # ── FIX-SYN-EXPAND: tabella sinonimi tecnici IT↔EN (15 cluster) ─────────────── # Struttura: ogni entry è un frozenset di termini equivalenti. # _expand_tokens() aggiunge tutti i sinonimi di ogni token del goal prima del matching. # Scelta design: sinonimi statici (zero LLM, zero latency) coprono l'80% dei task reali. # I cluster coprono i domini più frequenti nello sviluppo software. _SYN_CLUSTERS: list[frozenset[str]] = [ # Auth / Sicurezza frozenset({'auth', 'autenticazione', 'authentication', 'login', 'signin', 'guard', 'middleware', 'jwt', 'token', 'session', 'oauth', 'passport', 'credential', 'permission', 'role', 'accesso'}), # Pagamenti frozenset({'payment', 'pagamento', 'stripe', 'checkout', 'invoice', 'billing', 'subscription', 'abbonamento', 'fattura', 'webhook', 'price', 'plan', 'tier'}), # Database / ORM frozenset({'database', 'db', 'schema', 'model', 'migration', 'migrazione', 'orm', 'repository', 'query', 'drizzle', 'prisma', 'postgres', 'sqlite', 'mysql', 'table', 'tabella', 'record'}), # API / Network frozenset({'api', 'endpoint', 'route', 'rotta', 'router', 'server', 'request', 'response', 'richiesta', 'risposta', 'http', 'rest', 'graphql', 'fetch', 'axios', 'client'}), # UI / Frontend frozenset({'component', 'componente', 'ui', 'interface', 'interfaccia', 'button', 'form', 'modal', 'layout', 'page', 'pagina', 'style', 'css', 'theme', 'tema', 'render', 'view'}), # State Management frozenset({'state', 'stato', 'store', 'redux', 'zustand', 'context', 'provider', 'hook', 'reducer', 'action', 'dispatch', 'observable', 'signal', 'reactive'}), # File / Storage frozenset({'file', 'upload', 'caricamento', 'storage', 'bucket', 'download', 'attachment', 'allegato', 'blob', 'stream', 'filesystem', 'directory', 'path', 'percorso'}), # Testing frozenset({'test', 'testing', 'spec', 'unit', 'integration', 'e2e', 'mock', 'stub', 'fixture', 'assert', 'expect', 'coverage', 'vitest', 'jest', 'pytest'}), # Build / Deploy frozenset({'build', 'deploy', 'deployment', 'bundle', 'webpack', 'vite', 'esbuild', 'compile', 'dist', 'production', 'staging', 'pipeline', 'ci', 'cd', 'docker', 'container'}), # Email / Notifiche frozenset({'email', 'mail', 'smtp', 'notification', 'notifica', 'alert', 'push', 'telegram', 'slack', 'webhook', 'message', 'messaggio', 'sendgrid', 'resend', 'mailer'}), # AI / ML frozenset({'ai', 'llm', 'model', 'prompt', 'embedding', 'rag', 'vector', 'semantic', 'chat', 'completion', 'inference', 'openai', 'gemini', 'groq', 'anthropic', 'agent', 'agente'}), # Errori / Debug frozenset({'error', 'errore', 'exception', 'eccezione', 'bug', 'fix', 'debug', 'log', 'logging', 'trace', 'stack', 'crash', 'fallback', 'retry', 'recover', 'handler', 'catch'}), # Configurazione frozenset({'config', 'configurazione', 'configuration', 'settings', 'impostazioni', 'env', 'environment', 'variable', 'variabile', 'secret', 'segreto', 'dotenv', 'constant', 'costante'}), # Performance / Cache frozenset({'cache', 'performance', 'performanza', 'speed', 'velocità', 'optimize', 'ottimizzazione', 'lazy', 'memo', 'debounce', 'throttle', 'batch', 'compress', 'compressione'}), # Sicurezza / Validazione frozenset({'validation', 'validazione', 'validate', 'sanitize', 'sanitizzazione', 'schema', 'zod', 'yup', 'joi', 'csrf', 'xss', 'injection', 'escape', 'secure'}), # Monitoring / Observability (B-GAP-D: cluster mancante — task metriche/dashboard non rankati) frozenset({'metrics', 'metric', 'monitoring', 'monitoraggio', 'observability', 'prometheus', 'grafana', 'dashboard', 'telemetry', 'telemetria', 'tracing', 'trace', 'health', 'healthcheck', 'uptime', 'alerting', 'datadog', 'sentry', 'newrelic', 'audit', 'report'}), # Scheduling / Background Jobs (B-GAP-D: cluster mancante — task cron/queue/worker) frozenset({'cron', 'scheduler', 'pianificatore', 'schedule', 'queue', 'coda', 'worker', 'job', 'background', 'celery', 'bull', 'bullmq', 'agenda', 'delayed', 'periodic', 'retry', 'backoff', 'redis', 'task', 'processo', 'process', 'daemon'}), # WebSocket / Realtime (B-GAP-D: cluster mancante — task ws/sse/pubsub) frozenset({'websocket', 'ws', 'socket', 'socketio', 'realtime', 'real_time', 'sse', 'server_sent', 'pubsub', 'publish', 'subscribe', 'broadcast', 'channel', 'canale', 'room', 'event', 'listener', 'emitter', 'live', 'push', 'poll', 'long_polling', 'signalr', 'liveview'}), ] # Indice inverso: token → frozenset di sinonimi (costruito una volta a import) _SYN_INDEX: dict[str, frozenset[str]] = {} for _cluster in _SYN_CLUSTERS: for _term in _cluster: _SYN_INDEX[_term] = _cluster def _expand_tokens(tokens: list[str]) -> list[str]: """ FIX-SYN-EXPAND: espande ogni token del goal con i sinonimi IT/EN del suo cluster. Esempio: ["autenticazione", "aggiungi"] → ["autenticazione", "aggiungi", "auth", "login", "guard", "middleware", "jwt", ...] Garanzie: - Ordine stabile: token originali prima, sinonimi dopo (preserva priorità) - Nessun duplicato (usa set interno) - Nessun token < 3 chars, nessuna stopword aggiunta - Zero latency (<0.1ms per 20 token), zero LLM calls - Mai rilancia eccezioni """ try: seen: set[str] = set(tokens) expanded = list(tokens) for t in tokens: cluster = _SYN_INDEX.get(t) if cluster: for syn in cluster: if syn not in seen and len(syn) >= 3 and syn not in _RANK_STOP: seen.add(syn) expanded.append(syn) return expanded except Exception: return tokens def _split_camel_snake(text: str) -> list[str]: """ Spezza camelCase/PascalCase/snake_case in token lowercase (min 3 chars). Esempi: "contextManager" → ["context", "manager"] "rank_files_by_relevance" → ["rank", "files", "relevance"] "UnifiedAgentLoop" → ["unified", "agent", "loop"] Usato per fuzzy prefix bonus in rank_files_by_relevance. """ try: snake = _CAMEL_SPLIT_RE.sub(r'\1_\2', text) parts = re.split(r'[_\-./]', snake) return [p.lower() for p in parts if len(p) >= 3] except Exception: return [] def rank_files_by_relevance( goal: str, all_files: list[dict[str, Any]], k: int = 5, min_score: float = 0.0, ) -> list[str]: """ FIX-SKEL-RAG + FIX-SYN-EXPAND: Seleziona i top-K file più rilevanti per il goal. Score composito (normalizzato su max(len(base_tokens), 1)): path_hits * 2.0 — keyword del goal (espansi) nel path symbol_hits * 1.5 — keyword nei nomi funzione/classe (skeleton RAG) content_hits * 1.0 — keyword nei primi 600 chars del contenuto prefix_bonus * 0.4 — goal token è prefisso di un split-token path/symbol (fuzzy) entry_boost +0.15 — file entry-point/config noti lang_boost +0.20 — il linguaggio del file è nel goal FIX-SYN-EXPAND: - I token del goal vengono espansi con sinonimi IT/EN prima del matching. - Normalizzazione su len(base_tokens) originali (non espansi) per evitare score inflazionati su file che matchano solo sinonimi lontani. - "autenticazione" → matcha authGuard.ts, middleware.ts, jwt.ts anche senza keyword nel path — copertura semantica senza embeddings. Ritorna lista di path ordinata score-desc (top-K, score > min_score). Mai rilancia eccezioni — fallback ai primi K file non ranked. """ if not all_files or not goal: return [] try: base_tokens = [ t.lower() for t in re.findall(r'\b\w{3,}\b', goal[:500]) if t.lower() not in _RANK_STOP ] if not base_tokens: return [f.get('path', '') for f in all_files[:k] if f.get('path')] # FIX-SYN-EXPAND: espandi con sinonimi tecnici IT/EN tokens = _expand_tokens(base_tokens) goal_lower = goal.lower() # Normalizzatore: usa len(base_tokens) non len(tokens) per evitare score inflazionati n = max(len(base_tokens), 1) scores: list[tuple[float, str]] = [] for f in all_files: path = f.get('path', '') or '' content = (f.get('content', '') or '')[:600] lang = (f.get('language', '') or '').lower() if not path: continue path_lower = path.lower() content_lower = content.lower() # FIX-SKEL-RAG: estrai firme funzione/classe sigs = _extract_signatures(content, lang) symbols_lower = ' '.join(s.split(':', 1)[-1].lower() for s in sigs) # Score primario — matching su token espansi path_hits = sum(1 for t in tokens if t in path_lower) symbol_hits = sum(1 for t in tokens if t in symbols_lower) content_hits = sum(1 for t in tokens if t in content_lower) score = (path_hits * 2.0 + symbol_hits * 1.5 + content_hits) / n # Fuzzy prefix bonus (su token base, non espansi — evita falsi positivi) filename_stem = re.sub(r'\.[^.]+$', '', path_lower.rsplit('/', 1)[-1]) split_path = _split_camel_snake(filename_stem) split_syms = [t for s in sigs for t in _split_camel_snake(s.split(':', 1)[-1])] all_split = split_path + split_syms prefix_hits = sum( 1 for gt in base_tokens # usa base_tokens: fuzzy su originali for st in all_split if st != gt and st.startswith(gt) ) if prefix_hits: score += (prefix_hits * 0.4) / n # Entry-point boost if filename_stem in _RANK_ENTRY_STEMS: score += 0.15 # Language boost if lang and lang in goal_lower: score += 0.20 if score > min_score: scores.append((score, path)) # Ordinamento stabile: score desc, poi path asc scores.sort(key=lambda x: (-x[0], x[1])) return [p for _, p in scores[:k] if p] except Exception: return [f.get('path', '') for f in all_files[:k] if f.get('path')] def _extract_signatures(content: str, language: str) -> list[str]: """Estrae nomi di funzioni/classi per lo skeleton.""" try: lang = (language or '').lower() sigs: list[str] = [] if lang in ('typescript', 'ts', 'tsx', 'javascript', 'js', 'jsx'): for m in _FUNC_RE.finditer(content): name = m.group(1) or m.group(2) if name: sigs.append(f'fn:{name}') for m in _CLASS_RE.finditer(content): sigs.append(f'class:{m.group(1)}') elif lang in ('python', 'py'): for m in _PY_DEF_RE.finditer(content): sigs.append(f'def:{m.group(1)}') for m in _PY_CLS_RE.finditer(content): sigs.append(f'class:{m.group(1)}') return sigs[:15] except Exception: return [] def build_file_skeleton(path: str, content: str, language: str) -> str: """Costruisce una riga skeleton per un singolo file.""" sigs = _extract_signatures(content, language) line_count = content.count('\n') + 1 sigs_str = ', '.join(sigs[:8]) if sigs else '(no symbols)' return f' {path} ({line_count}L): {sigs_str}' async def build_project_skeleton(files: list[dict[str, Any]]) -> str: """ Costruisce lo skeleton compatto da una lista di file VFS. Ogni dict ha: path, content, language. Ritorna stringa multiriga per iniezione nel contesto agente. """ if not files: return '' try: lines = [f'\U0001f4c1 PROJECT SKELETON ({len(files)} files):'] for f in sorted(files, key=lambda x: x.get('path', '')): path = f.get('path', '?') content = f.get('content', '') or '' language = f.get('language', '') or '' lines.append(build_file_skeleton(path, content, language)) return '\n'.join(lines) except Exception: return '' async def compress_cold_file(path: str, content: str, language: str, tester_llm: Any | None = None) -> str: """ Comprime un file 'cold' al suo riepilogo essenziale. Usa Groq-8b via CONTEXT role per velocità. Fallback a skeleton. Max 8s. Mai rilancia eccezioni. """ # S573: hash() è PYTHONHASHSEED-salted → chiave diversa a ogni restart HF Space # → nessun riuso della cache tra restart. hashlib.sha256 è stabile e deterministica. _h = hashlib.sha256(content[:500].encode("utf-8", errors="replace")).hexdigest()[:16] cache_key = f'{path}:{_h}' if cache_key in _SUMMARY_CACHE: return _SUMMARY_CACHE[cache_key] skeleton = build_file_skeleton(path, content, language) if tester_llm and len(content) > 500: try: msgs = [ {"role": "system", "content": "Riassumi il file in max 2 righe: scopo, symbols chiave, deps. " "Solo facts. Formato: [SCOPO] | [SYMBOLS] | [DEPS]"}, {"role": "user", "content": f"File: {path}\n```{language}\n{content[:2000]}\n```"}, ] summary = await asyncio.wait_for( # S587: 120→200 — formato [SCOPO]|[SYMBOLS]|[DEPS] può superare 120 tok tester_llm.chat(msgs, temperature=0.0, max_tokens=200), timeout=7.0, ) if summary and not summary.startswith('[LLM'): result = f' {path}: {summary[:300]}' # S604: 180→300 — summary file LLM spesso 2-3 righe if len(_SUMMARY_CACHE) >= _MAX_SUMMARY_CACHE: oldest = next(iter(_SUMMARY_CACHE)) del _SUMMARY_CACHE[oldest] _SUMMARY_CACHE[cache_key] = result return result except Exception: pass # S364: fallback a skeleton return skeleton async def get_context_for_goal( goal: str, active_files: list[str], all_files: list[dict[str, Any]], tester_llm: Any | None = None, top_k: int = 5, ) -> str: """ Contesto intelligente per l'agente: - File in active_files: full content (max 1500 chars ciascuno) - Altri file: skeleton compatto - Output max: ~4000 chars S752-A + FIX-SKEL-RAG + FIX-SYN-EXPAND: se active_files è vuoto o None, usa rank_files_by_relevance() (con synonym expansion) per selezionare i top_k file più rilevanti per il goal. File con score == 0 esclusi automaticamente. """ if not all_files: return '' try: if not active_files and goal: active_files = rank_files_by_relevance(goal, all_files, k=top_k) active_set = set(active_files) parts: list[str] = [] budget = 4000 for f in all_files: path = f.get('path', '') if path not in active_set: continue content = (f.get('content', '') or '')[:1500] language = f.get('language', '') or '' chunk = f'[ACTIVE FILE: {path}]\n```{language}\n{content}\n```' parts.append(chunk) budget -= len(chunk) if budget <= 0: break cold_files = [f for f in all_files if f.get('path', '') not in active_set] if cold_files and budget > 500: skeleton = await build_project_skeleton(cold_files) if skeleton: parts.append(skeleton) return '\n\n'.join(parts) if parts else '' except Exception: return ''