sync: 113 file da Baida98/AI@8b66af87 (2026-06-21 20:19 UTC)

api/exec.py

@@ -521,3 +521,45 @@ async def llm_fix_code(
 
     return {'fixed_code': None, 'error': f'Tutti i provider falliti. Ultimo: {last_err[:300]}'}  # S588: 200→300
 
+
+# ── GAP-2: /api/exec/tool — dispatcher generico per tool calls dal frontend ────
+# Il frontend (toolExecutor.ts) chiama questo endpoint per tool come scaffold_project
+# che non hanno un endpoint dedicato. Il dispatcher risolve il nome tool nel TOOL_REGISTRY
+# e invoca la funzione associata (_fn) con gli args forniti.
+class ToolDispatchRequest(BaseModel):
+    tool: str
+    args: dict = {}
+
+
+@router.post('/api/exec/tool')
+async def exec_tool_dispatch(req: ToolDispatchRequest, request: Request):
+    """GAP-2: dispatcher generico — risolve tool nel TOOL_REGISTRY e chiama _fn."""
+    _internal_token = os.getenv('INTERNAL_TOKEN', '')
+    if _internal_token and request.headers.get('X-Internal-Token') != _internal_token:
+        raise HTTPException(401, 'Unauthorized')
+    try:
+        from tools.registry import TOOL_REGISTRY  # import locale — evita circular import
+    except ImportError as _ie:
+        return {'ok': False, 'error': f'TOOL_REGISTRY non disponibile: {_ie}'}
+    tool_def = TOOL_REGISTRY.get(req.tool)
+    if not tool_def:
+        available = ', '.join(list(TOOL_REGISTRY.keys())[:20])
+        return {'ok': False, 'error': f"Tool '{req.tool}' non trovato. Disponibili: {available}"}
+    _fn = tool_def.get('_fn')
+    if not _fn:
+        return {'ok': False, 'error': f"Tool '{req.tool}' non ha handler (_fn) — non eseguibile via dispatcher"}
+    try:
+        import asyncio as _asyncio
+        if _asyncio.iscoroutinefunction(_fn):
+            result = await _fn(**req.args)
+        else:
+            result = _fn(**req.args)
+        return {'ok': True, 'tool': req.tool, 'result': result}
+    except TypeError as _te:
+        # Parametri sbagliati — mostra la firma corretta
+        import inspect as _inspect
+        _sig = str(_inspect.signature(_fn))
+        return {'ok': False, 'error': f"Parametri non validi per '{req.tool}'{_sig}: {str(_te)[:200]}"}
+    except Exception as _e:
+        _logger.exception('[exec/tool] errore in %s', req.tool)
+        return {'ok': False, 'error': str(_e)[:500]}

tools/ast_check.py

@@ -0,0 +1,140 @@
+"""
+ast_check.py — P45: AST syntax check pre-esecuzione codice.
+
+Usa ast.parse() stdlib Python (zero dipendenze) per Python.
+Per JS/TS: esprima se disponibile, fallback balanced-bracket heuristic.
+Integra P17-B1: sostituisce routing regex E2B per validazione sintattica.
+
+Utilizzo:
+    from tools.ast_check import check_code_syntax
+    result = check_code_syntax(code, "python")
+    if not result.ok:
+        return f"SyntaxError: {result.error}"
+"""
+import ast
+import re
+import logging
+from dataclasses import dataclass
+
+_logger = logging.getLogger("tools.ast_check")
+
+
+@dataclass
+class SyntaxCheckResult:
+    ok: bool
+    error: str | None = None
+    language: str = ""
+    line: int | None = None
+    col: int | None = None
+
+    def __str__(self):
+        if self.ok:
+            return f"[AST] {self.language}: OK"
+        loc = f" (line {self.line})" if self.line else ""
+        return f"[AST] {self.language}{loc}: {self.error}"
+
+
+def check_python_syntax(code):
+    """AST check Python con ast.parse() stdlib. Zero dipendenze."""
+    try:
+        ast.parse(code, mode="exec")
+        return SyntaxCheckResult(ok=True, language="python")
+    except SyntaxError as e:
+        return SyntaxCheckResult(
+            ok=False, language="python",
+            error=f"{type(e).__name__}: {e.msg}",
+            line=e.lineno, col=e.offset,
+        )
+    except Exception as e:
+        return SyntaxCheckResult(ok=False, language="python", error=str(e))
+
+
+def _balanced_brackets(code):
+    """Heuristica bilanciamento parentesi — fallback JS senza esprima."""
+    stack = []
+    pairs = {")": "(", "}": "{", "]": "["}
+    in_str = None
+    i = 0
+    while i < len(code):
+        ch = code[i]
+        if in_str:
+            if ch == "\\" and i + 1 < len(code):
+                i += 2
+                continue
+            if ch == in_str:
+                in_str = None
+        elif ch in ('"', "'", '`'):
+            in_str = ch
+        elif ch in "({[":
+            stack.append(ch)
+        elif ch in ")}]":
+            if not stack or stack[-1] != pairs[ch]:
+                return SyntaxCheckResult(ok=False, language="js/ts", error=f"Unmatched '{ch}' at pos {i}")
+            stack.pop()
+        i += 1
+    if stack:
+        return SyntaxCheckResult(ok=False, language="js/ts", error=f"Unclosed '{stack[-1]}'")
+    return SyntaxCheckResult(ok=True, language="js/ts")
+
+
+def check_js_syntax(code):
+    """AST check JS/TS: esprima se disponibile, fallback balanced-bracket."""
+    try:
+        import esprima
+        esprima.parseScript(code, tolerant=False)
+        return SyntaxCheckResult(ok=True, language="js/ts")
+    except ImportError:
+        pass
+    except Exception as e:
+        msg = str(e)
+        m = re.search(r"Line (\d+)", msg)
+        line = int(m.group(1)) if m else None
+        return SyntaxCheckResult(ok=False, language="js/ts", error=msg[:200], line=line)
+    return _balanced_brackets(code)
+
+
+_LANG_MAP = {
+    "python": "python", "py": "python",
+    "javascript": "js", "js": "js", "jsx": "js",
+    "typescript": "js", "ts": "js", "tsx": "js",
+}
+
+
+def check_code_syntax(code, language):
+    """
+    Dispatch AST check per linguaggio. Fail-open su linguaggi non supportati.
+    Returns SyntaxCheckResult(ok, error, language, line, col).
+    """
+    if not code or not code.strip():
+        return SyntaxCheckResult(ok=True, language=language)
+    lang = _LANG_MAP.get(language.lower().strip(), "")
+    if lang == "python":
+        result = check_python_syntax(code)
+    elif lang == "js":
+        result = check_js_syntax(code)
+    else:
+        return SyntaxCheckResult(ok=True, language=language)
+    if not result.ok:
+        _logger.warning("[AST-P45] %s syntax error: %s", language, result.error)
+    return result
+
+
+def extract_code_blocks(text):
+    """Estrae blocchi codice da markdown. Ritorna list[(language, code)]."""
+    blocks = []
+    pattern = re.compile(r'```(\w+)?\n([\s\S]*?)```')
+    for m in pattern.finditer(text):
+        lang = (m.group(1) or "").strip()
+        code = m.group(2)
+        blocks.append((lang, code))
+    return blocks
+
+
+def check_all_code_blocks(text):
+    """Verifica tutti i blocchi codice in markdown. Ritorna solo errori."""
+    errors = []
+    for lang, code in extract_code_blocks(text):
+        result = check_code_syntax(code, lang)
+        if not result.ok:
+            errors.append(result)
+    return errors

tools/registry.py

@@ -1853,7 +1853,7 @@ async def _jina_fetch(
         max_length=max_length,
     )
 
-async def _python_analyze(code: str, filename: str = "<string>") -> dict:
+async def _python_analyze(code: str = "", content: str = "", filename: str = "<string>") -> dict:
     """P30-B1: Analisi statica Python in-process — zero exec_engine, zero deps esterne.
 
     Usa ast.parse() + visitor per:
@@ -1862,13 +1862,18 @@ async def _python_analyze(code: str, filename: str = "<string>") -> dict:
     - Suggerimenti actionable (funzioni troppo lunghe, nesting alto, ecc.)
 
     Zero I/O, zero network. Tipicamente <5ms.
+
+    GAP-1: accetta sia 'code' che 'content' — alias per compatibilità frontend.
+    Il frontend (toolDefsCode.ts) invia 'content', il backend usava solo 'code'.
     """
+    # GAP-1: alias — frontend può inviare 'content' invece di 'code'
+    code = code or content
     import ast as _ast
 
     result: dict = {"syntax_ok": False, "errors": [], "complexity": {}, "suggestions": [], "summary": ""}
 
     if not code or not code.strip():
-        result["errors"] = [{"type": "EmptyCode", "message": "Nessun codice fornito", "line": 0}]
+        result["errors"] = [{"type": "EmptyCode", "message": "Nessun codice fornito (parametro 'code' o 'content' richiesto)", "line": 0}]
         result["summary"] = "Codice vuoto"
         return result
 
@@ -2574,7 +2579,7 @@ TOOL_REGISTRY: dict[str, dict] = {
             "ottenere metriche strutturali, suggerire refactoring."
         ),
         "required_inputs": ["code"],
-        "optional_inputs": {"filename": "<string>"},
+        "optional_inputs": {"filename": "<string>", "content": ""},  # GAP-1: alias per 'code' (frontend compat)
         "risk_level": "low",
         "fallbacks": [],
         "_fn": _python_analyze,

tools/semantic_nav.py

@@ -0,0 +1,154 @@
+"""
+semantic_nav.py — Semantic Anchor Scoring (SAS)
+
+Scoring a 3 fattori:
+  1. Keyword Match   (peso 0.50) — intent semantico agente
+  2. Visual Position (peso 0.30) — struttura HTML semantica
+  3. Path Depth      (peso 0.20) — profondita URL
+
+auto_keywords_from_url(): fallback automatico se nessun intent esplicito.
+Backward-compatible: fallback regex se beautifulsoup4 non disponibile.
+"""
+import re
+import logging
+from urllib.parse import urljoin, urlparse
+
+_logger = logging.getLogger("tools.semantic_nav")
+
+KEYWORD_MATCH_WEIGHT   = 0.50
+VISUAL_POSITION_WEIGHT = 0.30
+PATH_DEPTH_WEIGHT      = 0.20
+
+VISUAL_POSITION_SCORES = {
+    "nav":     1.0,
+    "header":  0.8,
+    "main":    0.7,
+    "article": 0.6,
+    "section": 0.5,
+    "form":    0.4,
+    "aside":   0.3,
+    "footer":  0.1,
+    "default": 0.4,
+}
+
+_SKIP = re.compile(
+    r"(twitter\.com|x\.com|facebook\.com|instagram\.com|linkedin\.com|"
+    r"youtube\.com|cdn\.|analytics|pixel|tracking|googletagmanager|doubleclick|"
+    r"javascript:|mailto:|tel:|#$)",
+    re.I,
+)
+
+
+def auto_keywords_from_url(url):
+    """Estrae keywords dall URL come fallback intent."""
+    parsed = urlparse(url)
+    parts = []
+    host_parts = parsed.netloc.replace("www.", "").split(".")
+    parts.extend(host_parts[:-1])
+    path_parts = [s for s in parsed.path.split("/") if s and len(s) > 2]
+    parts.extend(path_parts)
+    for kv in (parsed.query or "").split("&"):
+        if "=" in kv:
+            k, v = kv.split("=", 1)
+            parts.extend([k, v])
+    _stop = {"the", "a", "an", "and", "or", "in", "on", "at", "to", "for", "of", "with"}
+    return [
+        p.lower().strip("-_")
+        for p in parts
+        if p and not p.isdigit() and p.lower() not in _stop and len(p) > 1
+    ][:12]
+
+
+def _get_parent_tag(tag):
+    current = tag
+    while current is not None:
+        name = getattr(current, "name", None)
+        if name in VISUAL_POSITION_SCORES:
+            return name
+        current = getattr(current, "parent", None)
+    return "default"
+
+
+def _score_link(text, url, tag, keywords):
+    score = 0.0
+    text_lower = text.lower()
+    for kw in keywords:
+        if kw.lower() in text_lower:
+            score += KEYWORD_MATCH_WEIGHT * 0.6
+            break
+    parent = getattr(tag, "parent", None)
+    ctx = parent.get_text(separator=" ", strip=True)[:300] if parent else ""
+    for kw in keywords:
+        if kw.lower() in ctx.lower():
+            score += KEYWORD_MATCH_WEIGHT * 0.4
+            break
+    parent_tag = _get_parent_tag(tag)
+    score += VISUAL_POSITION_SCORES.get(parent_tag, VISUAL_POSITION_SCORES["default"]) * VISUAL_POSITION_WEIGHT
+    path_segs = [s for s in urlparse(url).path.split("/") if s]
+    score += min(len(path_segs) / 5.0, 1.0) * PATH_DEPTH_WEIGHT
+    return max(0.0, min(score, 1.0))
+
+
+def _fallback_extract(html, base_url, keywords, max_links):
+    """Regex fallback quando bs4 non disponibile."""
+    base_domain = urlparse(base_url).netloc
+    links = []
+    kw_lower = [k.lower() for k in keywords]
+    for m in re.finditer(r'<a[^>]+href=["\']([ ^"\'>#][^"\']*)["\'\'][^>]*>([^<]{2,80})</a>', html, re.I):
+        raw_href, text = m.group(1).strip(), m.group(2).strip()
+        if not raw_href or _SKIP.search(raw_href):
+            continue
+        full_url = urljoin(base_url, raw_href)
+        if not full_url.startswith(("http://", "https://")):
+            continue
+        url_lower = full_url.lower()
+        kw_hit = any(kw in text.lower() or kw in url_lower for kw in kw_lower)
+        same_domain = urlparse(full_url).netloc == base_domain
+        score = (0.5 if kw_hit else 0.2) + (0.2 if same_domain else 0.0)
+        links.append({"text": text[:60], "url": full_url, "score": round(score, 2), "context": ""})
+    links.sort(key=lambda x: x["score"], reverse=True)
+    return links[:max_links]
+
+
+def extract_and_score_links(html_content, base_url, navigation_intent_keywords=None, max_links=10):
+    """
+    Estrae e prioritizza link con Semantic Anchor Scoring.
+    Returns: list[dict] con keys text, url, score, context. Ordinati per score.
+    """
+    keywords = navigation_intent_keywords or auto_keywords_from_url(base_url)
+    if not keywords:
+        keywords = ["main", "content", "article"]
+
+    try:
+        from bs4 import BeautifulSoup
+    except ImportError:
+        _logger.debug("[SAS] beautifulsoup4 non disponibile — uso fallback regex")
+        return _fallback_extract(html_content, base_url, keywords, max_links)
+
+    soup = BeautifulSoup(html_content, "html.parser")
+    results = []
+
+    for a_tag in soup.find_all("a", href=True):
+        text = a_tag.get_text(strip=True)
+        raw_href = a_tag["href"]
+        if not raw_href or _SKIP.search(raw_href):
+            continue
+        full_url = urljoin(base_url, raw_href)
+        if not full_url.startswith(("http://", "https://")):
+            continue
+        score = _score_link(text, full_url, a_tag, keywords)
+        ctx = ""
+        parent = a_tag.parent
+        if parent:
+            parent_text = parent.get_text(separator=" ", strip=True)
+            idx = parent_text.find(text)
+            if idx != -1:
+                ctx = parent_text[max(0, idx - 80):idx + len(text) + 80].strip()
+            else:
+                ctx = parent_text[:160].strip()
+        results.append({"text": text[:80], "url": full_url, "score": round(score, 3), "context": ctx[:200]})
+
+    results.sort(key=lambda x: x["score"], reverse=True)
+    top = results[:max_links]
+    _logger.debug("[SAS] %s: %d scored, top %d | kw=%s", base_url, len(results), len(top), keywords[:3])
+    return top

tools/web_fetch.py

@@ -30,6 +30,15 @@ from typing import Optional
 import logging
 _logger = logging.getLogger("tools.web_fetch")
 
+
+# SAS (P45-SAS): Semantic Anchor Scoring — lazy import, fail-safe fallback
+_sas_available = False
+try:
+    from tools import semantic_nav as _semantic_nav  # noqa: PLC0415
+    _sas_available = True
+except ImportError:
+    pass
+
 USER_AGENT = "Mozilla/5.0 (compatible; AgenteAI/3.0)"
 MAX_CONTENT = 6000          # W-NAV: alzato da 5000 → 6000 (allineato a MAX_TEXT browser.py)
 _PLAYWRIGHT_THRESHOLD = 300 # chars: se content < soglia → prova Playwright
@@ -197,7 +206,7 @@ async def _fetch_with_playwright(url: str, max_chars: int = MAX_CONTENT) -> Opti
 
 # ─── Endpoint principale ──────────────────────────────────────────────────────
 
-async def fetch_page(url: str, max_chars: int = MAX_CONTENT) -> dict:
+async def fetch_page(url: str, max_chars: int = MAX_CONTENT, navigation_intent_keywords: list[str] | None = None) -> dict:
     """
     Fetch + estrazione testo con pipeline a 3 livelli:
       L1: httpx statico + trafilatura (veloce, nessun overhead)
@@ -261,7 +270,18 @@ async def fetch_page(url: str, max_chars: int = MAX_CONTENT) -> dict:
                 result["extractor"] = "playwright"
             if warning:
                 result["warning"] = warning
-            result["relevant_links"] = _extract_relevant_links(html, url)
+            # SAS: prioritizza link per intent semantico (P45-SAS)
+            if _sas_available:
+                try:
+                    result["relevant_links"] = _semantic_nav.extract_and_score_links(
+                        html_content=html,
+                        base_url=url,
+                        navigation_intent_keywords=navigation_intent_keywords,
+                    )
+                except Exception:
+                    result["relevant_links"] = _extract_relevant_links(html, url)
+            else:
+                result["relevant_links"] = _extract_relevant_links(html, url)
             return result
 
     except httpx.TimeoutException: