""" 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 import logging from typing import Any _logger = logging.getLogger("agente_ai.context_manager") _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) ─────────────── _SYN_CLUSTERS: list[frozenset[str]] = [ frozenset({'auth', 'autenticazione', 'authentication', 'login', 'signin', 'guard', 'middleware', 'jwt', 'token', 'session', 'oauth', 'passport', 'credential', 'permission', 'role', 'accesso'}), frozenset({'payment', 'pagamento', 'stripe', 'checkout', 'invoice', 'billing', 'subscription', 'abbonamento', 'fattura', 'webhook', 'price', 'plan', 'tier'}), frozenset({'database', 'db', 'schema', 'model', 'migration', 'migrazione', 'orm', 'repository', 'query', 'drizzle', 'prisma', 'postgres', 'sqlite', 'mysql', 'table', 'tabella', 'record'}), frozenset({'api', 'endpoint', 'route', 'rotta', 'router', 'server', 'request', 'response', 'richiesta', 'risposta', 'http', 'rest', 'graphql', 'fetch', 'axios', 'client'}), frozenset({'component', 'componente', 'ui', 'interface', 'interfaccia', 'button', 'form', 'modal', 'layout', 'page', 'pagina', 'style', 'css', 'theme', 'tema', 'render', 'view'}), frozenset({'state', 'stato', 'store', 'redux', 'zustand', 'context', 'provider', 'hook', 'reducer', 'action', 'dispatch', 'observable', 'signal', 'reactive'}), frozenset({'file', 'upload', 'caricamento', 'storage', 'bucket', 'download', 'attachment', 'allegato', 'blob', 'stream', 'filesystem', 'directory', 'path', 'percorso'}), frozenset({'test', 'testing', 'spec', 'unit', 'integration', 'e2e', 'mock', 'stub', 'fixture', 'assert', 'expect', 'coverage', 'vitest', 'jest', 'pytest'}), frozenset({'build', 'deploy', 'deployment', 'bundle', 'webpack', 'vite', 'esbuild', 'compile', 'dist', 'production', 'staging', 'pipeline', 'ci', 'cd', 'docker', 'container'}), frozenset({'email', 'mail', 'smtp', 'notification', 'notifica', 'alert', 'push', 'telegram', 'slack', 'webhook', 'message', 'messaggio', 'sendgrid', 'resend', 'mailer'}), frozenset({'ai', 'llm', 'model', 'prompt', 'embedding', 'rag', 'vector', 'semantic', 'chat', 'completion', 'inference', 'openai', 'gemini', 'groq', 'anthropic', 'agent', 'agente'}), frozenset({'error', 'errore', 'exception', 'eccezione', 'bug', 'fix', 'debug', 'log', 'logging', 'trace', 'stack', 'crash', 'fallback', 'retry', 'recover', 'handler', 'catch'}), frozenset({'config', 'configurazione', 'configuration', 'settings', 'impostazioni', 'env', 'environment', 'variable', 'variabile', 'secret', 'segreto', 'dotenv', 'constant', 'costante'}), frozenset({'cache', 'performance', 'performanza', 'speed', 'velocità', 'optimize', 'ottimizzazione', 'lazy', 'memo', 'debounce', 'throttle', 'batch', 'compress', 'compressione'}), frozenset({'validation', 'validazione', 'validate', 'sanitize', 'sanitizzazione', 'schema', 'zod', 'yup', 'joi', 'csrf', 'xss', 'injection', 'escape', 'secure'}), frozenset({'metrics', 'metric', 'monitoring', 'monitoraggio', 'observability', 'prometheus', 'grafana', 'dashboard', 'telemetry', 'telemetria', 'tracing', 'trace', 'health', 'healthcheck', 'uptime', 'alerting', 'datadog', 'sentry', 'newrelic', 'audit', 'report'}), frozenset({'cron', 'scheduler', 'pianificatore', 'schedule', 'queue', 'coda', 'worker', 'job', 'background', 'celery', 'bull', 'bullmq', 'agenda', 'delayed', 'periodic', 'retry', 'backoff', 'redis', 'task', 'processo', 'process', 'daemon'}), 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'}), ] _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]: 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 as e: _logger.debug("[context_manager] _expand_tokens failed: %s", e) return tokens def _split_camel_snake(text: str) -> list[str]: 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 as e: _logger.debug("[context_manager] _split_camel_snake failed: %s", e) return [] def rank_files_by_relevance( goal: str, all_files: list[dict[str, Any]], k: int = 5, min_score: float = 0.0, ) -> list[str]: 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')] tokens = _expand_tokens(base_tokens) goal_lower = goal.lower() 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() sigs = _extract_signatures(content, lang) symbols_lower = ' '.join(s.split(':', 1)[-1].lower() for s in sigs) 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 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 for st in all_split if st != gt and st.startswith(gt) ) if prefix_hits: score += (prefix_hits * 0.4) / n if filename_stem in _RANK_ENTRY_STEMS: score += 0.15 if lang and lang in goal_lower: score += 0.20 if score > min_score: scores.append((score, path)) scores.sort(key=lambda x: (-x[0], x[1])) return [p for _, p in scores[:k] if p] except Exception as e: _logger.error("[context_manager] rank_files_by_relevance critical failure: %s", e) return [f.get('path', '') for f in all_files[:k] if f.get('path')] def _extract_signatures(content: str, language: str) -> list[str]: 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 as e: _logger.debug("[context_manager] _extract_signatures failed: %s", e) return [] def build_file_skeleton(path: str, content: str, language: str) -> str: 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: 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 as e: _logger.error("[context_manager] build_project_skeleton failed: %s", e) return ""