Spaces:
Running
Running
File size: 5,772 Bytes
28a08e7 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 | """
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
|