api/project_manifest.py

"""
project_manifest.py — ProjectManifest: multi-file coherence (S364)

Traccia esportazioni/importazioni per file VFS per dare all'agente
consapevolezza di quali file sono impattati quando uno cambia.

Design:
- Update fire-and-forget (mai blocca SSE)
- Silent failures always
- Regex-based, in-memory cache TTL 5min
"""
from __future__ import annotations
import logging
import re, time, logging
from typing import Any

_manifest_logger = logging.getLogger("agente_ai")

_MANIFEST_CACHE: dict[str, tuple[float, dict]] = {}
_CACHE_TTL = 300  # 5 min

_EXPORT_RE    = re.compile(
    r'export\s+(?:default\s+)?(?:function|class|const|let|var|type|interface|enum)\s+(\w+)',
    re.MULTILINE)
_IMPORT_RE    = re.compile(
    r"""import\s+.*?from\s+['"](.*?)['"]""",
    re.MULTILINE)
_PY_EXPORT_RE = re.compile(r'^(?:class|def)\s+([A-Za-z_]\w*)', re.MULTILINE)
_PY_IMPORT_RE = re.compile(r'^(?:from\s+(\S+)\s+import|import\s+(\S+))', re.MULTILINE)


def _extract_exports(content: str, language: str) -> list[str]:
    try:
        lang = (language or '').lower()
        if lang in ('typescript', 'ts', 'tsx', 'javascript', 'js', 'jsx'):
            return _EXPORT_RE.findall(content)[:20]
        elif lang in ('python', 'py'):
            return _PY_EXPORT_RE.findall(content)[:20]
    except Exception as _e:
        _manifest_logger.debug("project_manifest extract: %s", _e)
    return []


def _extract_imports(content: str, language: str) -> list[str]:
    try:
        lang = (language or '').lower()
        if lang in ('typescript', 'ts', 'tsx', 'javascript', 'js', 'jsx'):
            return _IMPORT_RE.findall(content)[:20]
        elif lang in ('python', 'py'):
            matches = _PY_IMPORT_RE.findall(content)
            return [m[0] or m[1] for m in matches if m[0] or m[1]][:20]
    except Exception as _e:
        _manifest_logger.debug("project_manifest extract: %s", _e)
    return []


def _get_cached(conversation_id: str) -> dict | None:
    entry = _MANIFEST_CACHE.get(conversation_id)
    if entry and (time.time() - entry[0]) < _CACHE_TTL:
        return entry[1]
    return None


def _set_cached(conversation_id: str, manifest: dict) -> None:
    _MANIFEST_CACHE[conversation_id] = (time.time(), manifest)
    if len(_MANIFEST_CACHE) > 50:
        oldest = min(_MANIFEST_CACHE.keys(), key=lambda k: _MANIFEST_CACHE[k][0])
        del _MANIFEST_CACHE[oldest]


async def update_manifest(conversation_id: str, file_path: str,
                           content: str, language: str) -> None:
    """Aggiorna il manifest per un singolo file. Fire-and-forget safe."""
    if not conversation_id or not file_path:
        return
    try:
        manifest = _get_cached(conversation_id) or {}
        manifest[file_path] = {
            'exports':    _extract_exports(content, language),
            'imports':    _extract_imports(content, language),
            'lines':      content.count('\n') + 1,
            'updated_at': int(time.time()),
        }
        _set_cached(conversation_id, manifest)
    except Exception:
        pass  # S364: silent failure always


async def get_impacted_files(conversation_id: str, changed_path: str) -> list[str]:
    """
    Ritorna file path che importano da changed_path.
    Usato per avvertire l'agente dei file che potrebbero richiedere aggiornamenti.
    """
    try:
        manifest = _get_cached(conversation_id)
        if not manifest:
            return []
        base     = changed_path.lstrip('./').rsplit('.', 1)[0]
        impacted = []
        for path, info in manifest.items():
            if path == changed_path:
                continue
            for imp in info.get('imports', []):
                imp_norm = imp.lstrip('./').rsplit('.', 1)[0]
                if imp_norm == base or imp.endswith('/' + base.split('/')[-1]):
                    impacted.append(path)
                    break
        return impacted[:10]
    except Exception:
        return []


async def get_skeleton(conversation_id: str) -> str:
    """
    Ritorna lo skeleton compatto del progetto: path + exports.
    Fornisce all'agente consapevolezza del progetto senza contenuto completo.
    """
    try:
        manifest = _get_cached(conversation_id)
        if not manifest:
            return ''
        lines = ['=== PROJECT SKELETON ===']
        for path, info in sorted(manifest.items()):
            exports = info.get('exports', [])
            n_lines = info.get('lines', 0)
            exp_str = ', '.join(exports[:5]) if exports else '(no exports)'
            lines.append(f'\u2022 {path} ({n_lines}L) \u2192 {exp_str}')
        lines.append('=== END SKELETON ===')
        return '\n'.join(lines)
    except Exception:
        return ''


async def build_manifest_from_vfs(conversation_id: str) -> None:
    """
    Ricostruisce il manifest da tutti i file VFS. Chiamato lazy.
    Fire-and-forget.
    """
    try:
        from .state import sb
        data = sb().table('vfs_files').select(
            'path, content, language'
        ).eq('conversation_id', conversation_id).execute()
        files    = data.data or []
        manifest = {}
        for f in files:
            path     = f.get('path', '')
            content  = f.get('content', '') or ''
            language = f.get('language', '') or ''
            if path and content:
                manifest[path] = {
                    'exports':    _extract_exports(content, language),
                    'imports':    _extract_imports(content, language),
                    'lines':      content.count('\n') + 1,
                    'updated_at': int(time.time()),
                }
        if manifest:
            _set_cached(conversation_id, manifest)
    except Exception:
        pass  # S364: silent failure always