tools/registry.py

"""
registry.py — Tool Registry
Ogni tool ha: goal, required_inputs, risk_level, fallbacks, _fn.
"""
import httpx
import asyncio
import subprocess
import tempfile
import os
import sys
import ipaddress
import socket
from contextvars import ContextVar

import logging
_logger = logging.getLogger("tools.registry")

# ─── SEC-FS-JAIL: confinamento path per read_file/write_file/apply_patch ──
# GAP: pathlib.Path(path) usato as-is permetteva accesso a qualunque file
# leggibile/scrivibile dal processo (env mount, secrets, codice di altri
# servizi). Confina tutte le operazioni FS del tool registry sotto una root
# esplicita, configurabile via env (default: cwd del processo backend).
_FS_JAIL_ROOT = _pathlib_module = None
def _fs_jail_root():
    import pathlib as _pl
    global _FS_JAIL_ROOT
    if _FS_JAIL_ROOT is None:
        _root = os.getenv("FS_TOOL_ROOT", os.getcwd())
        _FS_JAIL_ROOT = _pl.Path(_root).resolve()
    return _FS_JAIL_ROOT

def _safe_fs_path(path: str):
    """Risolve 'path' e verifica che sia contenuto in _fs_jail_root().
    Ritorna (resolved_path, None) se ok, oppure (None, error_msg) se rifiutato.
    Blocca path assoluti fuori jail e traversal (../) che escono dalla root."""
    import pathlib as _pl
    try:
        _root = _fs_jail_root()
        _candidate = (_root / path) if not _pl.Path(path).is_absolute() else _pl.Path(path)
        _resolved = _candidate.resolve()
        _resolved.relative_to(_root)
        return _resolved, None
    except ValueError:
        return None, f"Accesso negato: path fuori dalla sandbox consentita ({path})"
    except Exception as exc:
        return None, f"Path non valido: {exc}"

# ─── SEC-SSRF: validazione IP per call_api ─────────────────────────────
# GAP: _call_api chiamava qualsiasi URL senza risolvere l'hostname o
# bloccare IP privati/loopback/link-local, permettendo di colpire endpoint
# interni del backend stesso (es. http://localhost:PORT/api/vault/...).
_SSRF_ALLOWED_SCHEMES = frozenset({"http", "https"})

def _ssrf_check(url: str):
    """Ritorna None se l'URL è sicuro da chiamare, stringa di errore se bloccato."""
    from urllib.parse import urlparse
    try:
        _parsed = urlparse(url)
    except Exception as exc:
        return f"URL non valido: {exc}"
    if _parsed.scheme.lower() not in _SSRF_ALLOWED_SCHEMES:
        return f"Scheme non permesso: '{_parsed.scheme}' (solo http/https)"
    _host = _parsed.hostname or ""
    if not _host:
        return "URL senza hostname valido"
    if _host.lower() in ("localhost", "0.0.0.0", "::1"):
        return "SSRF bloccato: hostname locale non permesso"
    try:
        _ip = ipaddress.ip_address(socket.gethostbyname(_host))
        if _ip.is_private or _ip.is_loopback or _ip.is_link_local or _ip.is_reserved or _ip.is_multicast:
            return "SSRF bloccato: IP privato/loopback/riservato non permesso"
    except socket.gaierror:
        return f"Hostname non risolvibile: {_host}"
    except Exception:
        pass  # errori di parsing IP imprevisti: lascia proseguire, httpx gestirà l'errore di rete
    return None

# ─── S749-GAP1/D: backend-exec microservice client + ContextVar session ──
_EXEC_ENGINE_URL: str = os.getenv("EXEC_ENGINE_URL", "").rstrip("/")
_EXEC_TOKEN_HDR: str  = os.getenv("EXEC_TOKEN", "")

# S749-D: ContextVar per session_id isolata per task — thread-safe, asyncio-safe.
# unified_loop.run() imposta questo valore al boot del task e lo resetta nel finally.
# Default "agent_default" se chiamato fuori dal loop (es. test unitari).
_agent_session_id_var: ContextVar[str] = ContextVar("agent_session_id", default="agent_default")

# Shell validation -> tools._shell_safety
from tools._shell_safety import (
    validate_shell_command as _validate_shell_cmd,
    run_shell_safe_sync as _run_shell_sync,
    safe_shell_env as _safe_env,
)
import re as _re_registry
_REGISTRY_SAFE_CMD_RE = _re_registry.compile(
    r'^(ls|cat|echo|pwd|whoami|date|uname|python3?|node|npm|pip[3]?|pnpm|'
    r'grep|find|head|tail|wc|sort|uniq|diff|mkdir|touch|cp|mv|chmod|'
    r'git\s+(status|log|diff|show)|curl\s+https?://|wget\s+https?://)(\s|$)',
    _re_registry.IGNORECASE,
)
_REGISTRY_SHELL_METACHAR_RE = _re_registry.compile(r'[;&|`<>\n\r]|\$[\(\{]')

# P17-B3: persistent HTTP client per backend-exec — evita nuovo TCP/TLS handshake ad ogni call.
# Risparmio stimato: ~100-300ms per chiamata run_python/execute_shell su Railway.
# Reset automatico su connessione persa (is_closed check in _get_exec_client).
_exec_http_client: "httpx.AsyncClient | None" = None

def _get_exec_client() -> "httpx.AsyncClient":
    """Lazy singleton httpx.AsyncClient con keepalive per backend-exec."""
    global _exec_http_client
    if _exec_http_client is None or _exec_http_client.is_closed:
        _exec_http_client = httpx.AsyncClient(
            timeout=httpx.Timeout(20.0, connect=5.0),
            limits=httpx.Limits(
                max_connections=5,
                max_keepalive_connections=3,
                keepalive_expiry=60.0,
            ),
        )
    return _exec_http_client

async def _call_exec_engine(payload: dict, endpoint: str = "/api/exec") -> "dict | None":
    """Chiama backend-exec microservice (Railway) con fallback silente.

    Ritorna il JSON della risposta se status 200, altrimenti None.
    Se EXEC_ENGINE_URL non configurato ritorna None immediatamente.
    Timeout conservativo 20s — mai blocca il loop principale.
    S750-GAP-A: 1 retry con back-off 2s — gestisce cold start Railway (~1-2s attesa).
    """
    if not _EXEC_ENGINE_URL:
        return None
    headers: dict = {}
    if _EXEC_TOKEN_HDR:
        headers["X-Exec-Token"] = _EXEC_TOKEN_HDR
    for _attempt in range(2):  # S750-GAP-A: tentativi 0 e 1
        try:
            _c = _get_exec_client()  # P17-B3: persistent client — reusa connessione TCP
            _r = await _c.post(
                f"{_EXEC_ENGINE_URL}{endpoint}",
                json=payload,
                headers=headers,
            )
            if _r.status_code == 200:
                return _r.json()
        except (httpx.TransportError, httpx.ConnectError, ConnectionError, OSError) as _exc:
            _logger.debug("[registry] P17-B3 transport error, resetting client: %s", type(_exc).__name__)  # noqa: BLE001
            global _exec_http_client
            _exec_http_client = None  # P17-B3: reset SOLO su errori trasporto/rete
        except Exception as _exc:
            _logger.debug("[registry] P17-B3 silenced (no reset): %s", type(_exc).__name__)  # noqa: BLE001
        if _attempt == 0:
            await asyncio.sleep(2.0)  # back-off 2s prima del retry
    return None


# ─── Tool functions ────────────────────────────────────────

async def _web_search(query: str, max_results: int = 5) -> dict:
    """
    S357: Parallelismo completo — tutti e 4 i provider lanciati con asyncio.gather.
    Worst case: 10s (timeout singolo provider) invece di 40s+ (4 × 10s sequenziali).
    Merge con deduplicazione per URL, priorità Brave > Tavily > Wikipedia > DDG.
    """
    import re as _re, html as _html, urllib.parse as _urlparse, urllib.request as _urlreq
    _headers = {"User-Agent": "Mozilla/5.0 (compatible; AgentBot/1.0)"}
    brave_key  = os.environ.get("BRAVE_SEARCH_API_KEY", "")
    tavily_key = os.environ.get("TAVILY_API_KEY", "")

    async def _brave() -> list:
        if not brave_key:
            return []
        try:
            async with httpx.AsyncClient(timeout=10) as c:
                r = await c.get(
                    "https://api.search.brave.com/res/v1/web/search",
                    params={"q": query, "count": max_results, "text_decorations": "0"},
                    headers={**_headers, "Accept": "application/json", "X-Subscription-Token": brave_key},
                )
                if r.status_code == 200:
                    return [
                        # S602: snippet 300→500 — Brave description[:300] parity con Tavily
                        {"title": it["title"], "snippet": (it.get("description") or "")[:500],
                         "url": it["url"], "source": "Brave"}
                        for it in r.json().get("web", {}).get("results", [])[:max_results]
                        if it.get("title") and it.get("url")
                    ]
        except Exception as _exc:
            _logger.debug("[registry] silenced %s", type(_exc).__name__)  # noqa: BLE001
        return []

    async def _tavily() -> list:
        if not tavily_key:
            return []
        try:
            async with httpx.AsyncClient(timeout=10) as c:
                r = await c.post(
                    "https://api.tavily.com/search",
                    json={"api_key": tavily_key, "query": query,
                          "max_results": max_results, "include_answer": False},
                    headers={**_headers, "Content-Type": "application/json"},
                )
                if r.status_code == 200:
                    return [
                        # S600: snippet 300→500 — Tavily restituisce snippet più ricchi
                    {"title": it["title"], "snippet": (it.get("content") or "")[:500],
                         "url": it["url"], "source": "Tavily"}
                        for it in r.json().get("results", [])[:max_results]
                        if it.get("title") and it.get("url")
                    ]
        except Exception as _exc:
            _logger.debug("[registry] silenced %s", type(_exc).__name__)  # noqa: BLE001
        return []

    async def _wikipedia() -> list:
        try:
            wiki_qs = _urlparse.urlencode({
                "action": "query", "list": "search", "srsearch": query,
                "format": "json", "utf8": "1", "srlimit": min(max_results, 4), "srnamespace": "0",
            })
            wiki_req = _urlreq.Request(
                f"https://en.wikipedia.org/w/api.php?{wiki_qs}",
                headers={"User-Agent": "agente-ai/3.2"},
            )
            wiki_data = await asyncio.to_thread(
                lambda: __import__("json").loads(_urlreq.urlopen(wiki_req, timeout=7).read())
            )
            return [
                {
                    "title":   item.get("title", ""),
                    "url":     "https://en.wikipedia.org/wiki/" + item.get("title", "").replace(" ", "_"),
                    # S600: snippet 300→500 — Wikipedia snippet può essere più lungo
                    "snippet": _html.unescape(_re.sub(r"<[^>]+>", "", item.get("snippet", "")))[:500],
                    "source":  "Wikipedia",
                }
                for item in wiki_data.get("query", {}).get("search", [])[:max_results]
            ]
        except Exception:
            return []

    async def _ddg() -> list:
        try:
            ddg_qs  = _urlparse.urlencode({"q": query, "kl": "it-it"})
            ddg_req = _urlreq.Request(
                f"https://html.duckduckgo.com/html/?{ddg_qs}",
                headers={
                    "User-Agent": "Mozilla/5.0 (iPhone; CPU iPhone OS 17_0 like Mac OS X) AppleWebKit/605.1.15 Safari/604.1",
                    "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
                    "Accept-Language": "it-IT,it;q=0.9,en;q=0.8",
                },
            )
            raw = await asyncio.to_thread(
                lambda: _urlreq.urlopen(ddg_req, timeout=10).read().decode("utf-8", errors="replace")
            )
            links    = _re.findall(r'class="result__a"[^>]+href="([^"]+)"[^>]*>(.*?)</a>', raw, _re.S)
            snippets = _re.findall(r'class="result__snippet"[^>]*>(.*?)</a>', raw, _re.S)
            out = []
            for i, (href, title_html) in enumerate(links[:max_results]):
                title   = _html.unescape(_re.sub(r"<[^>]+>", "", title_html)).strip()
                snippet = _html.unescape(_re.sub(r"<[^>]+>", "", snippets[i] if i < len(snippets) else "")).strip()
                if title and href.startswith("http"):
                    # S600: snippet 300→500 — DDG snippet spesso viene troncato a 300
                    out.append({"title": title, "url": href, "snippet": snippet[:500], "source": "DDG"})
            return out
        except Exception:
            return []

    async def _jina() -> list:
        """S378: Jina Search — gratuito, no API key, fallback affidabile."""
        try:
            import urllib.parse as _up
            jina_url = f"https://s.jina.ai/?q={_up.quote(query)}"
            async with httpx.AsyncClient(timeout=10) as c:
                r = await c.get(
                    jina_url,
                    headers={**_headers, "Accept": "application/json", "X-No-Cache": "true"},
                )
                if r.status_code == 200:
                    data = r.json().get("data", [])
                    return [
                        {
                            "title":   item.get("title", ""),
                            "url":     item.get("url", ""),
                            # S602: snippet 300→500 — Jina description/content parity con altri provider
                            "snippet": (item.get("description") or item.get("content") or "")[:500],
                            "source":  "Jina",
                        }
                        for item in data[:max_results]
                        if item.get("title") and item.get("url")
                    ]
        except Exception as _exc:
            _logger.debug("[registry] silenced %s", type(_exc).__name__)  # noqa: BLE001
        return []

    async def _hackernews() -> list:
        """S763: HackerNews via Algolia — gratuito, no API key, qualità alta su query tech."""
        try:
            async with httpx.AsyncClient(timeout=8) as c:
                r = await c.get(
                    "https://hn.algolia.com/api/v1/search",
                    params={"query": query, "hitsPerPage": min(max_results, 4), "tags": "story"},
                )
                if r.status_code == 200:
                    return [
                        {"title": h.get("title", "")[:300],  # S607: 200→300
                         "snippet": (h.get("story_text") or "")[:300],  # S583: 250→300
                         "url": h.get("url") or f"https://news.ycombinator.com/item?id={h.get('objectID','')}",
                         "source": "HackerNews"}
                        for h in r.json().get("hits", [])
                        if h.get("title")
                    ]
        except Exception as _exc:
            _logger.debug("[registry] silenced %s", type(_exc).__name__)  # noqa: BLE001
        return []

    # S357: lancio parallelo — worst case = max timeout singolo (10s), non 4×10s=40s
    # S378: Jina aggiunto come 5° provider (gratuito, no API key)
    # GAP3-fix: HackerNews aggiunto come 6° provider (tech quality, da web_search.py ora integrato)
    _gather_results = await asyncio.gather(
        _brave(), _tavily(), _wikipedia(), _ddg(), _jina(), _hackernews(),
        return_exceptions=True,
    )
    all_lists = [r for r in _gather_results if not isinstance(r, BaseException)]

    # Merge con deduplicazione per URL (priorità: Brave > Tavily > Wikipedia > DDG > Jina)
    seen_urls: set = set()
    results: list = []
    for res_list in all_lists:
        for item in (res_list if isinstance(res_list, list) else []):
            url = item.get("url", "")
            if url and url not in seen_urls:
                seen_urls.add(url)
                results.append(item)
                if len(results) >= max_results:
                    break
        if len(results) >= max_results:
            break

    return {"query": query, "results": results[:max_results]}

async def _read_page(url: str, query: str = "") -> dict:
    """S763: upgrade — usa extract_with_trafilatura (Readability-quality).
    Fallback a regex strip se trafilatura non disponibile.
    query opzionale per filtrare paragrafi rilevanti.
    """
    try:
        async with httpx.AsyncClient(timeout=15, follow_redirects=True) as c:
            r = await c.get(
                url,
                headers={"User-Agent": "Mozilla/5.0 (compatible; AgentBot/1.0)"},
            )
            try:
                from tools.content_cleaner import extract_with_trafilatura
                result = extract_with_trafilatura(
                    html=r.text, url=url, query=query or None, max_chars=5000,
                )
                return {
                    "url":       url,
                    "content":   result["content"],
                    "status":    r.status_code,
                    "extractor": result.get("extractor", "trafilatura"),
                    "chars":     result.get("chars", 0),
                }
            except ImportError:
                import re as _re
                _c = _re.sub(r"<[^>]+>", " ", r.text)
                _c = _re.sub(r"\s+", " ", _c).strip()
                return {"url": url, "content": _c[:5000], "status": r.status_code, "extractor": "regex"}
    except Exception as e:
        return {"url": url, "content": "", "error": str(e)[:300]}

async def _get_weather(city: str) -> dict:
    try:
        async with httpx.AsyncClient(timeout=8) as c:
            # S390-B-H: usa params= invece di f-string per URL encoding corretto.
            # f-string non codifica spazi/accenti → API restituisce 0 risultati per "New York", "Reggio Emilia", ecc.
            geo = await c.get(
                "https://geocoding-api.open-meteo.com/v1/search",
                params={"name": city, "count": 1, "language": "it", "format": "json"},
            )
            results = geo.json().get("results", [])
            if not results:
                return {"error": f"Città '{city}' non trovata"}
            loc = results[0]
            weather = await c.get(
                f"https://api.open-meteo.com/v1/forecast?latitude={loc['latitude']}&longitude={loc['longitude']}"
                f"&current=temperature_2m,weather_code,wind_speed_10m&timezone=auto"
            )
            w = weather.json().get("current", {})
            return {"city": loc["name"], "country": loc.get("country", ""), "temp_c": w.get("temperature_2m"), "wind_kmh": w.get("wind_speed_10m"), "code": w.get("weather_code")}
    except Exception as e:
        return {"error": str(e)}

async def _calculate(expression: str) -> dict:
    try:
        import ast, operator
        # S390-B-M: aggiunto Mod (%) e FloorDiv (//) agli operator consentiti
        allowed = {ast.Add: operator.add, ast.Sub: operator.sub, ast.Mult: operator.mul, ast.Div: operator.truediv, ast.Pow: operator.pow, ast.USub: operator.neg, ast.Mod: operator.mod, ast.FloorDiv: operator.floordiv}
        def eval_node(node):
            if isinstance(node, ast.Constant): return node.value
            if isinstance(node, ast.BinOp): return allowed[type(node.op)](eval_node(node.left), eval_node(node.right))
            if isinstance(node, ast.UnaryOp): return allowed[type(node.op)](eval_node(node.operand))
            raise ValueError("Operazione non supportata")
        tree = ast.parse(expression, mode='eval')
        result = eval_node(tree.body)
        return {"expression": expression, "result": result}
    except Exception as e:
        return {"expression": expression, "error": str(e)}

async def _run_python(code: str) -> dict:
    """S574/S749: run_python con backend-exec microservice (sandbox persistente per sessione).

    Priorità:
    1. backend-exec (EXEC_ENGINE_URL configurato) — sandbox persistente, pip install per sessione,
       resource limits 512 MB, TTL cleaner. Usa AGENT_SESSION_ID env come session_id.
    2. Fallback: exec_sandbox locale (S574) — tmpdir effimera, comportamento pre-S749.

    Vantaggio chiave: se lo step precedente ha installato pandas, questo step lo trova.
    """
    # S749-D: session_id da ContextVar (impostata da unified_loop per task isolation)
    _session_id = _agent_session_id_var.get()
    _remote = await _call_exec_engine({
        "code": code,
        "language": "python",
        "timeout": 15,
        "session_id": _session_id,
    })
    if _remote is not None:
        # Normalizza formato: backend-exec → {exit_code, stdout, stderr}
        #                      exec_sandbox → {returncode, stdout, stderr}
        return {
            "returncode": _remote.get("exit_code", -1),
            "stdout":     _remote.get("stdout", ""),
            "stderr":     _remote.get("stderr", ""),
        }
    # Fallback locale (S574)
    from api.exec_sandbox import run_in_sandbox_async
    return await run_in_sandbox_async(code, lang="python",
                                      task_id=_session_id, timeout=15.0)  # GAP-2: usa session_id per task isolation (no race condition)



async def _generate_image(prompt: str, width: int = 512, height: int = 512) -> dict:
    """
    Genera immagine AI: FLUX.1-schnell via backend /api/vision/generate (HF Space, gratuito),
    fallback Pollinations AI se FLUX non disponibile.
    """
    import urllib.parse

    # V001: Prova prima il backend interno FLUX (stesso processo, localhost)
    _base_url = os.environ.get("BACKEND_BASE_URL", "http://localhost:7860")
    _token    = os.environ.get("INTERNAL_TOKEN", "")
    try:
        async with httpx.AsyncClient(timeout=30.0) as c:
            _r = await c.post(
                f"{_base_url}/api/vision/generate",
                json={"prompt": prompt.strip()[:600], "width": width, "height": height},
                headers={**({"X-Internal-Token": _token} if _token else {})},
            )
            if _r.status_code == 200:
                _data = _r.json()
                if _data.get("ok") and _data.get("image_url"):
                    return {
                        "url":    _data["image_url"],
                        "prompt": prompt,
                        "width":  width,
                        "height": height,
                        "ready":  True,
                        "source": "flux",
                        "note":   "Immagine generata con FLUX.1-schnell via backend.",
                    }
    except Exception:
        pass  # FLUX non raggiungibile → Pollinations fallback

    # Fallback: Pollinations AI (gratuito, nessuna API key)
    encoded = urllib.parse.quote(prompt.strip(), safe="")
    seed    = sum(ord(c) for c in prompt) % 9999 + 1
    url = (
        f"https://image.pollinations.ai/prompt/{encoded}"
        f"?width={width}&height={height}&seed={seed}&nologo=true&enhance=true"
    )
    return {
        "url":    url,
        "prompt": prompt,
        "width":  width,
        "height": height,
        "ready":  True,
        "source": "pollinations",
        "note":   "Copia l\'URL nel browser o incollalo in un tag <img> per vedere l\'immagine.",
    }

async def _browser_navigate(url: str, wait_ms: int = 2000, mobile: bool = False) -> dict:
    """
    S403 — Limite 1 (criticità 10/10): Browser execution layer.
    Naviga a una URL con Playwright headless, restituisce titolo + testo + DOM links/inputs.
    Stateless (nessuna sessione persistente) — usa per lettura/scraping rapido.
    Playwright è già installato nel backend (browser.py v5 S65).
    """
    try:
        from api.browser import (
            _safe_url, _LAUNCH_ARGS, _make_context, _DOM_SCRIPT,
            GOTO_TIMEOUT, MAX_LINKS, MAX_INPUTS, MAX_TEXT
        )
        if not _safe_url(url):
            return {"error": "URL non consentita (localhost/intranet bloccato per sicurezza)"}
        from playwright.async_api import async_playwright
        async with async_playwright() as pw:
            browser = await pw.chromium.launch(headless=True, args=_LAUNCH_ARGS)
            ctx     = await _make_context(browser, 1280, 800, mobile)
            page    = await ctx.new_page()
            try:
                await page.goto(url, wait_until="domcontentloaded", timeout=GOTO_TIMEOUT)
                await page.wait_for_timeout(wait_ms)
                title   = await page.title()
                dom_raw = await page.evaluate(_DOM_SCRIPT % (MAX_LINKS, MAX_INPUTS, MAX_TEXT))
                text    = (dom_raw.get("text") or "")[:2000] if isinstance(dom_raw, dict) else ""
                links   = (dom_raw.get("links") or [])[:15] if isinstance(dom_raw, dict) else []
                inputs  = (dom_raw.get("inputs") or [])[:10] if isinstance(dom_raw, dict) else []
                return {
                    "url":    page.url,
                    "title":  title,
                    "text":   text,
                    "links":  links,
                    "inputs": inputs,
                }
            except Exception as e:
                # S600: 300→500 — browser inner exception può contenere stack/path
                return {"url": url, "error": str(e)[:500]}
            finally:
                await ctx.close()
                await browser.close()
    except ImportError:
        return {"error": "Playwright non disponibile — usa read_page come alternativa"}
    except Exception as e:
        # S600: 300→500 — outer exception
        return {"error": str(e)[:500]}


async def _browser_session_open(url: str, wait_ms: int = 1500, mobile: bool = False) -> dict:
    """
    S403 — Apre sessione Playwright persistente (stateful).
    Restituisce session_id da usare con browser_session_act e browser_session_close.
    Max 2 sessioni simultanee (OOM guard HF free tier).
    Usa questa modalità per task multi-step: login → naviga → compila → invia.
    """
    try:
        from api.browser import (
            _safe_url, _LAUNCH_ARGS, _make_context, _DOM_SCRIPT,
            _sessions, SESSION_LIMIT, GOTO_TIMEOUT, MAX_LINKS, MAX_INPUTS, MAX_TEXT,
            _close_session
        )
        import uuid, time
        if not _safe_url(url):
            return {"error": "URL non consentita"}
        if len(_sessions) >= SESSION_LIMIT:
            oldest = min(_sessions, key=lambda sid: _sessions[sid]["last_used"])
            await _close_session(oldest, "OOM guard from tool")
        from playwright.async_api import async_playwright
        pw      = await async_playwright().start()
        browser = await pw.chromium.launch(headless=True, args=_LAUNCH_ARGS)
        ctx     = await _make_context(browser, 1280, 800, mobile)
        page    = await ctx.new_page()
        await page.goto(url, wait_until="domcontentloaded", timeout=GOTO_TIMEOUT)
        await page.wait_for_timeout(wait_ms)
        title   = await page.title()
        dom_raw = await page.evaluate(_DOM_SCRIPT % (MAX_LINKS, MAX_INPUTS, MAX_TEXT))
        sid     = uuid.uuid4().hex[:16]
        _sessions[sid] = {
            "pw": pw, "browser": browser, "context": ctx, "page": page,
            "created_at": time.time(), "last_used": time.time(), "url": page.url,
        }
        return {
            "session_id": sid,
            "url":        page.url,
            "title":      title,
            "text":       (dom_raw.get("text") or "")[:1500] if isinstance(dom_raw, dict) else "",
            "links":      (dom_raw.get("links") or [])[:15] if isinstance(dom_raw, dict) else [],
            "inputs":     (dom_raw.get("inputs") or [])[:10] if isinstance(dom_raw, dict) else [],
        }
    except ImportError:
        return {"error": "Playwright non disponibile"}
    except Exception as e:
        # S600: 300→500 — registry browser session exception
        return {"error": str(e)[:500]}


async def _browser_session_act(session_id: str, actions: list, wait_ms: int = 1000) -> dict:
    """
    S403 — Esegue azioni su sessione Playwright esistente.
    actions: lista di {type, selector?, value?, key?, ms?}
    Tipi supportati: click, fill, select, press, hover, wait_for, wait, scroll.
    Restituisce DOM aggiornato (titolo + testo + links + inputs).
    """
    try:
        from api.browser import (
            _sessions, _execute_actions, _DOM_SCRIPT,
            MAX_LINKS, MAX_INPUTS, MAX_TEXT
        )
        import time
        sess = _sessions.get(session_id)
        if not sess:
            return {"error": f"Sessione {session_id} non trovata o scaduta (TTL 8 min)"}
        sess["last_used"] = time.time()
        page = sess["page"]

        # Converti lista dict → oggetti con attributo .type, .selector, ecc.
        class _A:
            def __init__(self, d: dict):
                self.type     = d.get("type", "")
                self.selector = d.get("selector")
                self.value    = d.get("value")
                self.key      = d.get("key")
                self.ms       = d.get("ms")

        await _execute_actions(page, [_A(a) for a in (actions or [])])
        await page.wait_for_timeout(wait_ms)
        sess["url"] = page.url
        title   = await page.title()
        dom_raw = await page.evaluate(_DOM_SCRIPT % (MAX_LINKS, MAX_INPUTS, MAX_TEXT))
        return {
            "session_id": session_id,
            "url":        page.url,
            "title":      title,
            "text":       (dom_raw.get("text") or "")[:1500] if isinstance(dom_raw, dict) else "",
            "links":      (dom_raw.get("links") or [])[:15] if isinstance(dom_raw, dict) else [],
            "inputs":     (dom_raw.get("inputs") or [])[:10] if isinstance(dom_raw, dict) else [],
        }
    except ImportError:
        return {"error": "Playwright non disponibile"}
    except Exception as e:
        # S601: 300→500 — parity con altri session handler
        return {"error": str(e)[:500]}


async def _browser_session_close(session_id: str) -> dict:
    """S403 — Chiude sessione Playwright e libera memoria (~300 MB/sessione)."""
    try:
        from api.browser import _sessions, _close_session
        if session_id not in _sessions:
            return {"ok": True, "note": "Sessione già chiusa o non trovata"}
        await _close_session(session_id, "tool call")
        return {"ok": True, "session_id": session_id}
    except Exception as e:
        return {"ok": False, "error": str(e)[:300]}  # S589: 200→300


async def _get_news(query: str, max_results: int = 5) -> dict:
    """S378: get_news — alias di web_search con query ottimizzata per notizie recenti.
    smolagents chiama questo tool per richieste di tipo 'notizie su X'.
    """
    news_query = f"notizie recenti {query}" if not any(
        kw in query.lower() for kw in ("notizie", "news", "ultime", "latest", "breaking")
    ) else query
    return await _web_search(news_query, max_results=max_results)


# ─── S666: tool FS (apply_patch/write_file/read_file/execute_shell) ───────────
# Questi tool erano in unified_loop._TOOL_MAP (S659) ma NON in TOOL_REGISTRY.
# agent.py usa TOOL_REGISTRY per lookup → chiamate a questi tool falliscono con
# KeyError silenzioso. Aggiunte _fn + entries per copertura completa del registry.

import pathlib as _pathlib

async def _read_file(path: str, encoding: str = "utf-8") -> dict:
    """S666/SEC-FS-JAIL: Legge un file dal filesystem del backend, confinato a _fs_jail_root()."""
    try:
        _p, _err = _safe_fs_path(path)
        if _err:
            return {"ok": False, "error": _err}
        if not _p.exists():
            return {"ok": False, "error": f"File non trovato: {path}"}
        if _p.stat().st_size > 10 * 1024 * 1024:  # 10 MB limit
            return {"ok": False, "error": f"File troppo grande (>{10}MB): {path}"}
        text = _p.read_text(encoding=encoding, errors="replace")
        return {"ok": True, "content": text, "path": path, "size": len(text)}
    except PermissionError:
        return {"ok": False, "error": f"Accesso negato: {path}"}
    except Exception as exc:
        return {"ok": False, "error": str(exc)}


async def _write_file(path: str, content: str, encoding: str = "utf-8") -> dict:
    """S666/GAP-2/SEC-FS-JAIL: Scrive/sovrascrive un file nel filesystem del backend, confinato a _fs_jail_root().
    Manus Gap fix: read-back verifica completezza scrittura.
    write_text() puo completare senza eccezione anche su FS con truncation silenziosa."""
    try:
        _p, _err = _safe_fs_path(path)
        if _err:
            return {"ok": False, "error": _err}
        _p.parent.mkdir(parents=True, exist_ok=True)
        _p.write_text(content, encoding=encoding)
        # GAP-2: read-back — verifica che il file abbia la dimensione attesa
        # apply_patch fa gia questo (+ rollback TS); ora anche _write_file e atomico
        _written = _p.read_text(encoding=encoding)
        if len(_written) != len(content):
            return {
                "ok":    False,
                "error": f"Write truncated: expected {len(content)} chars, got {len(_written)}",
                "path":  path,
            }
        return {"ok": True, "path": path, "size": len(_written)}
    except PermissionError:
        return {"ok": False, "error": f"Accesso negato: {path}"}
    except Exception as exc:
        return {"ok": False, "error": str(exc)}


async def _ts_syntax_check(path: str) -> tuple[bool, str]:
    """GAP-6: verifica sintassi TypeScript via node --check dopo apply_patch.
    Ritorna (ok, error_msg). Non-bloccante: su node assente o timeout assume ok."""
    try:
        result = await asyncio.to_thread(
            subprocess.run,
            ["node", "--check", path],
            capture_output=True, text=True, timeout=10,
        )
        if result.returncode == 0:
            return True, ""
        return False, (result.stderr or result.stdout).strip()[:500]
    except (FileNotFoundError, subprocess.TimeoutExpired):
        return True, ""  # node non disponibile o timeout → assume ok (fail-safe)
    except Exception:
        return True, ""  # fail-safe assoluto — non deve mai bloccare il tool


async def _apply_patch(path: str, patch: str) -> dict:
    """S666/SEC-FS-JAIL: Applica una patch unified-diff a un file esistente, confinato a _fs_jail_root()."""
    import io
    try:
        _p, _err = _safe_fs_path(path)
        if _err:
            return {"ok": False, "error": _err}
        if not _p.exists():
            return {"ok": False, "error": f"File non trovato: {path}"}
        original = _p.read_text(encoding="utf-8", errors="replace")

        # Prova con subprocess patch(1) se disponibile, altrimenti Python fallback
        result = await asyncio.to_thread(
            subprocess.run,
            ["patch", "-p0", "--input=-", str(_p)],
            input=patch, capture_output=True, text=True, timeout=15,
        )
        if result.returncode == 0:
            # GAP-6: typecheck post-patch su .ts/.tsx — rollback automatico se sintassi invalida
            if path.endswith((".ts", ".tsx")):
                _ts_ok, _ts_err = await _ts_syntax_check(path)
                if not _ts_ok:
                    try: _p.write_text(original, encoding="utf-8")
                    except Exception: pass
                    return {"ok": False, "error": f"Rollback: sintassi TS invalida dopo patch — {_ts_err}", "path": path, "rolled_back": True}
            return {"ok": True, "path": path, "applied": True, "output": result.stdout.strip()}

        # Fallback: ricerca/sostituzione del blocco più semplice
        lines = patch.splitlines()
        removes = [l[1:] for l in lines if l.startswith("-") and not l.startswith("---")]
        adds    = [l[1:] for l in lines if l.startswith("+") and not l.startswith("+++")]
        if removes and len(removes) == len(adds):
            patched = original
            for rem, add in zip(removes, adds):
                patched = patched.replace(rem, add, 1)
            if patched != original:
                _p.write_text(patched, encoding="utf-8")
                # GAP-6: typecheck post-patch su .ts/.tsx — rollback automatico se sintassi invalida
                if path.endswith((".ts", ".tsx")):
                    _ts_ok, _ts_err = await _ts_syntax_check(path)
                    if not _ts_ok:
                        try: _p.write_text(original, encoding="utf-8")
                        except Exception: pass
                        return {"ok": False, "error": f"Rollback: sintassi TS invalida dopo patch (fallback) — {_ts_err}", "path": path, "rolled_back": True}
                return {"ok": True, "path": path, "applied": True, "method": "fallback_replace"}
        return {"ok": False, "error": result.stderr.strip() or "Patch non applicata", "path": path}
    except FileNotFoundError:
        return {"ok": False, "error": "patch(1) non trovato — solo fallback Python disponibile"}
    except Exception as exc:
        return {"ok": False, "error": str(exc)}


# S679/S749: execute_shell con backend-exec microservice (sessione persistente)
async def _execute_shell(command: str, timeout: int = 30, cwd: str = ".") -> dict:
    """S666/S749: Esegue un comando shell con timeout e cattura output.

    Priorità:
    1. backend-exec (EXEC_ENGINE_URL) — sessione persistente, pip install condiviso con run_python.
    2. Fallback: subprocess locale con asyncio.to_thread.
    """
    _shell_err = _validate_shell_cmd(command)
    if _shell_err:
        return {"ok":False,"error":_shell_err,"command":command}
    # S749-D: session_id da ContextVar (impostata da unified_loop per task isolation)
    _session_id = _agent_session_id_var.get()
    _remote = await _call_exec_engine({
        "command": command,
        "timeout": min(timeout, 60),
        "session_id": _session_id,
    }, endpoint="/api/execute-shell")
    if _remote is not None:
        _ec = _remote.get("exit_code", -1)
        return {
            "ok":      _ec == 0,
            "stdout":  _remote.get("stdout", "")[:8192],
            "stderr":  _remote.get("stderr", "")[:2048],
            "code":    _ec,
            "command": command,
        }
    # Fallback locale — _run_shell_sync (NO shell=True, env pulita)
    try:
        _fb = await asyncio.to_thread(_run_shell_sync, command, cwd, min(timeout,120))
        return {"ok":_fb["ok"],"stdout":_fb["stdout"],"stderr":_fb["stderr"],
                "code":_fb.get("code",-1),"command":command,
        }
    except subprocess.TimeoutExpired:
        return {"ok": False, "error": f"Timeout {timeout}s superato", "command": command}
    except Exception as exc:
        return {"ok": False, "error": str(exc), "command": command}


# ─── Registry ─────────────────────────────────────────────



# ── V002-V004: nuovi tool backend (web_research, send_email, database_query) ──

async def _web_research(topic: str, depth: int = 4, synthesize: bool = True) -> dict:
    """Ricerca web multi-fonte con sintesi AI via /api/web/research."""
    _base = os.environ.get("BACKEND_BASE_URL", "http://localhost:7860")
    _tok  = os.environ.get("INTERNAL_TOKEN", "")
    try:
        async with httpx.AsyncClient(timeout=60.0) as c:
            r = await c.post(
                f"{_base}/api/web/research",
                json={"topic": str(topic)[:400], "depth": min(int(depth), 8), "synthesize": synthesize},
                headers={**({"X-Internal-Token": _tok} if _tok else {})},
            )
            if r.status_code == 200:
                return r.json()
    except Exception as exc:
        return {"ok": False, "error": str(exc)[:200]}
    return {"ok": False, "error": "web_research endpoint non disponibile"}


async def _send_email(to: str, subject: str, body: str, html: bool = False, from_name: str = "Agente AI") -> dict:
    """Invia email via /api/email/send (richiede RESEND_API_KEY)."""
    _base = os.environ.get("BACKEND_BASE_URL", "http://localhost:7860")
    _tok  = os.environ.get("INTERNAL_TOKEN", "")
    try:
        async with httpx.AsyncClient(timeout=20.0) as c:
            r = await c.post(
                f"{_base}/api/email/send",
                json={"to": to, "subject": subject, "body": body, "html": html, "from_name": from_name},
                headers={**({"X-Internal-Token": _tok} if _tok else {})},
            )
            if r.status_code == 200:
                return r.json()
    except Exception as exc:
        return {"ok": False, "message": str(exc)[:200]}
    return {"ok": False, "message": "send_email endpoint non disponibile"}


async def _database_query(sql: str, params: list | None = None, read_only: bool = True) -> dict:
    """Esegue query SQL su DATABASE_URL configurato via /api/database/query."""
    _base = os.environ.get("BACKEND_BASE_URL", "http://localhost:7860")
    _tok  = os.environ.get("INTERNAL_TOKEN", "")
    try:
        async with httpx.AsyncClient(timeout=30.0) as c:
            r = await c.post(
                f"{_base}/api/database/query",
                json={"sql": str(sql)[:2000], "params": params or [], "read_only": read_only},
                headers={**({"X-Internal-Token": _tok} if _tok else {})},
            )
            if r.status_code == 200:
                return r.json()
    except Exception as exc:
        return {"ok": False, "error": str(exc)[:200]}
    return {"ok": False, "error": "database_query endpoint non disponibile"}



async def _call_api(url: str, method: str = "GET", headers: dict | None = None,
                    body=None, auth_type: str = "none", auth_value: str = "",
                    timeout_ms: int = 15000) -> dict:
    """S601: Chiama qualsiasi endpoint REST esterno con metodo/headers/body/auth configurabili.
    SEC-SSRF: valida scheme + risolve hostname, blocca IP privati/loopback/riservati.
    follow_redirects=False per evitare bypass SSRF via redirect verso IP interni."""
    import json as _j
    _ssrf_err = _ssrf_check(url)
    if _ssrf_err:
        return {"ok": False, "error": _ssrf_err, "url": url}
    _method = method.upper()
    _headers = dict(headers or {})
    _timeout = min(float(timeout_ms) / 1000, 30.0)
    # auth injection
    if auth_type == "bearer" and auth_value:
        _headers["Authorization"] = f"Bearer {auth_value}"
    elif auth_type == "basic" and auth_value:
        import base64 as _b64
        _headers["Authorization"] = "Basic " + _b64.b64encode(auth_value.encode()).decode()
    elif auth_type == "api_key" and auth_value and ":" in auth_value:
        k, v = auth_value.split(":", 1)
        _headers[k.strip()] = v.strip()
    try:
        async with httpx.AsyncClient(timeout=_timeout, follow_redirects=False) as c:
            kwargs = {"headers": _headers}
            if body is not None:
                if isinstance(body, (dict, list)):
                    kwargs["json"] = body
                else:
                    kwargs["content"] = str(body).encode()
                    _headers.setdefault("Content-Type", "text/plain")
            r = await c.request(_method, url, **kwargs)
            _ct = r.headers.get("content-type", "")
            if "json" in _ct:
                try:
                    resp_body = r.json()
                except Exception:
                    resp_body = r.text[:4000]
            else:
                resp_body = r.text[:4000]
            return {
                "ok": True, "status": r.status_code,
                "body": resp_body, "headers": dict(r.headers),
            }
    except Exception as exc:
        return {"ok": False, "status": 0, "error": str(exc)[:200]}


async def _execute_sql(sql: str, params: list | None = None, read_only: bool = True) -> dict:
    """S601: alias di _database_query — esegue SQL su DATABASE_URL (PostgreSQL/SQLite)."""
    return await _database_query(sql=sql, params=params, read_only=read_only)


async def _create_pdf(content: str, filename: str = "documento.pdf",
                      format: str = "html") -> dict:
    """S601: Genera PDF da HTML/Markdown/testo via backend /api/vision/pdf."""
    _base = os.environ.get("BACKEND_BASE_URL", "http://localhost:7860")
    _tok  = os.environ.get("INTERNAL_TOKEN", "")
    try:
        async with httpx.AsyncClient(timeout=30.0) as c:
            r = await c.post(
                f"{_base}/api/vision/pdf",
                json={"content": str(content)[:50000], "filename": filename, "format": format},
                headers={**({"X-Internal-Token": _tok} if _tok else {})},
            )
            if r.status_code == 200:
                return r.json()
            return {"ok": False, "error": f"HTTP {r.status_code}", "detail": r.text[:200]}
    except Exception as exc:
        return {"ok": False, "error": str(exc)[:200]}


# ─── S763: 10 tool mancanti (S760 dichiarati mai implementati) ──────────────
# Presenti in planner.py PLANNER_SYSTEM, agent.py _STEP_VISIBILITY/_NARR_QUICK,
# _TOOL_NEEDED_RE ma senza _fn in TOOL_REGISTRY -> KeyError silenzioso al runtime.

async def _directory_tree(path: str = ".", max_depth: int = 3, show_hidden: bool = False) -> dict:
    """S763: Albero filesystem con os.walk."""
    import os as _os
    try:
        base = _os.path.abspath(path)
        if not _os.path.isdir(base):
            return {"error": f"Percorso non trovato: {path}"}
        _IGNORE = {".git", "__pycache__", "node_modules", ".venv", "venv", "dist", "build", ".next", ".cache"}
        lines: list = [f"{base}/"]
        count = 0
        for root, dirs, files in _os.walk(base):
            depth = root.replace(base, "").count(_os.sep)
            if depth >= max_depth:
                dirs.clear()
                continue
            dirs[:] = sorted(d for d in dirs
                             if (show_hidden or not d.startswith(".")) and d not in _IGNORE)
            ind = "  " * (depth + 1)
            for d in dirs:
                lines.append(f"{ind}+-- {d}/")
            for fname in sorted(files):
                if not show_hidden and fname.startswith("."):
                    continue
                if count >= 200:
                    lines.append(f"{ind}... (troncato)")
                    break
                lines.append(f"{ind}    {fname}")
                count += 1
        return {"ok": True, "path": base, "tree": "\n".join(lines), "count": count}
    except Exception as e:
        return {"error": str(e)[:300]}


async def _file_search(pattern: str, path: str = ".", file_glob: str = "*") -> dict:
    """S763: grep -rn con fallback Python os.walk+read."""
    import os as _os
    try:
        proc = await asyncio.create_subprocess_exec(
            "grep", "-rn", "--include", file_glob, "--color=never", "-m", "5",
            pattern, _os.path.abspath(path),
            stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE,
        )
        out, _ = await asyncio.wait_for(proc.communicate(), timeout=15)
        lines = [l for l in out.decode("utf-8", errors="replace").splitlines() if l.strip()]
        if proc.returncode == 1 and not lines:
            return {"ok": True, "pattern": pattern, "matches": [], "count": 0, "note": "Nessun risultato"}
        return {"ok": True, "pattern": pattern, "path": path, "matches": lines[:50], "count": len(lines)}
    except (FileNotFoundError, asyncio.TimeoutError):
        import re as _re2
        matches: list = []
        try:
            _pat = _re2.compile(pattern, _re2.IGNORECASE)
            for root, _, files in _os.walk(path):
                for fname in files:
                    fpath = _os.path.join(root, fname)
                    try:
                        with open(fpath, "r", encoding="utf-8", errors="replace") as fh:
                            for i, line in enumerate(fh, 1):
                                if _pat.search(line):
                                    matches.append(f"{fpath}:{i}: {line.rstrip()[:200]}")
                                    if len(matches) >= 50:
                                        break
                    except Exception:
                        continue
                    if len(matches) >= 50:
                        break
        except Exception as ex:
            return {"error": str(ex)[:300]}
        return {"ok": True, "pattern": pattern, "matches": matches, "count": len(matches)}
    except Exception as e:
        return {"error": str(e)[:300]}


async def _git_status(cwd: str = ".") -> dict:
    """S763: branch + status --short + log -5. Read-only."""
    try:
        async def _rg(*args: str) -> str:
            p = await asyncio.create_subprocess_exec(
                "git", *args, cwd=cwd,
                stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE,
            )
            out, _ = await asyncio.wait_for(p.communicate(), timeout=10)
            return out.decode("utf-8", errors="replace").strip()
        return {
            "ok":     True,
            "branch": await _rg("rev-parse", "--abbrev-ref", "HEAD"),
            "status": await _rg("status", "--short") or "(working tree clean)",
            "log":    await _rg("log", "--oneline", "-5"),
        }
    except Exception as e:
        return {"error": str(e)[:300]}


async def _git_clone(url: str, directory: str = "", depth: int = 0) -> dict:
    """S763: git clone <url> [dir]. Timeout 120s. Risk medium."""
    try:
        cmd = ["git", "clone"]
        if depth > 0:
            cmd += ["--depth", str(depth)]
        cmd.append(url)
        if directory:
            cmd.append(directory)
        proc = await asyncio.create_subprocess_exec(
            *cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE,
        )
        out, err = await asyncio.wait_for(proc.communicate(), timeout=120)
        combined = (out + err).decode("utf-8", errors="replace")[:1000]
        if proc.returncode == 0:
            target = directory or url.rstrip("/").split("/")[-1].removesuffix(".git")
            return {"ok": True, "directory": target, "output": combined}
        return {"ok": False, "error": combined}
    except asyncio.TimeoutError:
        return {"ok": False, "error": "timeout 120s"}
    except Exception as e:
        return {"ok": False, "error": str(e)[:300]}


async def _git_diff(cwd: str = ".", staged: bool = False) -> dict:
    """S763: git diff [--cached]. Stat + diff max 3000 chars. Read-only."""
    try:
        extra = ["--cached"] if staged else []
        async def _rg(*a: str) -> str:
            p = await asyncio.create_subprocess_exec(
                "git", *a, cwd=cwd,
                stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE,
            )
            out, _ = await asyncio.wait_for(p.communicate(), timeout=15)
            return out.decode("utf-8", errors="replace")
        stat = await _rg("diff", *extra, "--stat", "--no-color")
        diff = await _rg("diff", *extra, "--no-color")
        return {"staged": staged, "stat": stat[:1000], "diff": diff[:3000] or "(nessuna modifica)"}
    except Exception as e:
        return {"error": str(e)[:300]}




async def _recall(query: str, limit: int = 5) -> dict:
    """S-GAP13: cerca in agentMemory (chiave/valore in-process). Fallback su entries recenti."""
    try:
        from api.state import _get_mem_manager_async as _gmm
        mem = await _gmm()
        if mem is None:
            return {"results": [], "note": "MemoryManager non disponibile"}
        results = []
        try:
            raw = await mem.search(query, limit=limit)
            results = [{"key": r.get("key",""), "value": r.get("value",""), "score": r.get("score",0)} for r in (raw or [])]
        except Exception:
            try:
                raw = await mem.list(limit=limit * 2)
                q_lower = query.lower()
                for r in (raw or []):
                    k = str(r.get("key","")).lower()
                    v = str(r.get("value","")).lower()
                    if q_lower in k or q_lower in v:
                        results.append({"key": r.get("key",""), "value": r.get("value","")})
                        if len(results) >= limit:
                            break
            except Exception as _exc:
                _logger.debug("[registry] silenced %s", type(_exc).__name__)  # noqa: BLE001
        return {"results": results, "count": len(results), "query": query}
    except Exception as e:
        return {"results": [], "error": str(e)[:200]}


async def _list_files(path: str = ".", recursive: bool = False, max_items: int = 100) -> dict:
    """S-GAP13: elenca file nella directory. os.listdir/os.walk. Max 100 items."""
    import os as _os
    try:
        path = _os.path.abspath(path)
        if not _os.path.exists(path):
            return {"ok": False, "error": f"Path non trovato: {path}"}
        if not _os.path.isdir(path):
            return {"ok": False, "error": f"Non e una directory: {path}"}
        items = []
        if recursive:
            for root, dirs, files in _os.walk(path):
                dirs[:] = [d for d in dirs if not d.startswith('.') and d not in ('node_modules','__pycache__','.git')]
                rel_root = _os.path.relpath(root, path)
                for f in files:
                    rel = _os.path.join(rel_root, f) if rel_root != '.' else f
                    items.append(rel)
                    if len(items) >= max_items:
                        break
                if len(items) >= max_items:
                    break
        else:
            for entry in _os.scandir(path):
                items.append(entry.name + ('/' if entry.is_dir() else ''))
                if len(items) >= max_items:
                    break
        return {"ok": True, "path": path, "items": items, "count": len(items), "truncated": len(items) >= max_items}
    except Exception as e:
        return {"ok": False, "error": str(e)[:300]}


def _diff_text(text_a: str, text_b: str, context_lines: int = 3) -> dict:
    """S-GAP13: confronta due testi con difflib.unified_diff. Restituisce patch testo."""
    import difflib
    try:
        lines_a = text_a.splitlines(keepends=True)
        lines_b = text_b.splitlines(keepends=True)
        diff = list(difflib.unified_diff(lines_a, lines_b, fromfile="a", tofile="b", n=context_lines))
        patch = "".join(diff)
        added   = sum(1 for l in diff if l.startswith('+') and not l.startswith('+++'))
        removed = sum(1 for l in diff if l.startswith('-') and not l.startswith('---'))
        return {"patch": patch[:4000], "added": added, "removed": removed, "identical": len(diff) == 0}
    except Exception as e:
        return {"patch": "", "error": str(e)[:200]}


def _validate_json(json_str: str, schema: dict | None = None) -> dict:
    """S-GAP13: valida JSON (json.loads). Con schema dict usa jsonschema se disponibile."""
    import json
    try:
        parsed = json.loads(json_str)
        result: dict = {"valid": True, "type": type(parsed).__name__}
        if schema:
            try:
                import jsonschema
                jsonschema.validate(parsed, schema)
                result["schema_valid"] = True
            except ImportError:
                result["schema_note"] = "jsonschema non installato — validazione struttura skippata"
            except Exception as ve:
                result["valid"] = False
                result["schema_error"] = str(ve)[:400]
        return result
    except json.JSONDecodeError as e:
        return {"valid": False, "error": f"JSON non valido: {e.msg} (riga {e.lineno}, col {e.colno})"}
    except Exception as e:
        return {"valid": False, "error": str(e)[:200]}


async def _lint_code_tool(content: str, language: str = "auto", path: str = "") -> dict:
    """S-GAP13: analisi statica codice. Wrapper di api.linter.lint_code. Auto-detect da estensione."""
    try:
        if language == "auto" and path:
            ext = path.rsplit(".", 1)[-1].lower() if "." in path else ""
            language = {"py": "python", "js": "javascript", "jsx": "javascript",
                        "ts": "typescript", "tsx": "typescript", "json": "json"}.get(ext, "python")
        from api.linter import lint_code as _lc
        return await _lc(content=content, language=language, path=path)
    except Exception as e:
        return {"ok": False, "errors": [], "warnings": [], "error": str(e)[:200]}

async def _git_push(remote: str = "origin", branch: str = "", cwd: str = ".") -> dict:
    """S-GAP12: git push <remote> [<branch>]. Timeout 70s. Risk high."""
    import asyncio as _aio
    try:
        cmd = ["git", "push", remote]
        if branch:
            cmd.append(branch)
        proc = await _aio.create_subprocess_exec(
            *cmd, cwd=cwd,
            stdout=_aio.subprocess.PIPE, stderr=_aio.subprocess.PIPE,
        )
        out, err = await _aio.wait_for(proc.communicate(), timeout=65)
        combined = (out + err).decode("utf-8", errors="replace")[:800]
        return {"ok": proc.returncode == 0, "output": combined, "code": proc.returncode}
    except _aio.TimeoutError:
        return {"ok": False, "error": "git push timeout (65s)"}
    except Exception as e:
        return {"ok": False, "error": str(e)[:300]}


async def _git_sync_vfs(
    files: dict,
    branch: str = "agent-state",
    message: str = "chore(vfs): auto-sync session",
    repo: str = "",
) -> dict:
    """
    RF-1: git_sync_vfs — Commit atomico VFS→GitHub via Git Data API.

    Flusso: GET HEAD → POST blob×N → POST tree → POST commit → PATCH/POST ref.
    Branch inesistente: creato automaticamente da HEAD di main.
    Repo: param repo oppure env GITHUB_REPO.
    Fail-safe: ritorna error se GITHUB_TOKEN mancante.
    """
    import base64
    import httpx as _httpx
    import os as _os

    gh_token = _os.environ.get("GITHUB_TOKEN", "")
    if not gh_token:
        return {"success": False, "error": "GITHUB_TOKEN non configurato"}
    if not files:
        return {"success": False, "error": "Nessun file da sincronizzare"}

    gh_repo = repo or _os.environ.get("GITHUB_REPO", "")
    if not gh_repo:
        return {"success": False, "error": "Specifica repo='owner/repo' oppure imposta GITHUB_REPO env"}

    _hdrs = {
        "Authorization": f"token {gh_token}",
        "Accept": "application/vnd.github.v3+json",
        "Content-Type": "application/json",
    }
    base_url = f"https://api.github.com/repos/{gh_repo}"

    try:
        async with _httpx.AsyncClient(timeout=30.0, headers=_hdrs) as _c:
            # 1. Leggi HEAD branch target (o fallback a main)
            base_sha: str | None = None
            branch_exists = False
            _ref_r = await _c.get(f"{base_url}/git/ref/heads/{branch}")
            if _ref_r.status_code == 200:
                base_sha = _ref_r.json()["object"]["sha"]
                branch_exists = True
            else:
                _main_r = await _c.get(f"{base_url}/git/ref/heads/main")
                if _main_r.status_code == 200:
                    base_sha = _main_r.json()["object"]["sha"]
                else:
                    return {"success": False, "error": f"Impossibile leggere HEAD: {_main_r.status_code}"}

            # 2. Crea blob per ogni file
            tree_items: list[dict] = []
            for fpath, content in files.items():
                if not isinstance(content, str):
                    content = str(content)
                encoded = base64.b64encode(content.encode("utf-8", errors="replace")).decode()
                _blob_r = await _c.post(f"{base_url}/git/blobs", json={"content": encoded, "encoding": "base64"})
                if _blob_r.status_code not in (200, 201):
                    return {"success": False, "error": f"Blob fail [{fpath}]: {_blob_r.status_code}"}
                tree_items.append({"path": fpath, "mode": "100644", "type": "blob", "sha": _blob_r.json()["sha"]})

            # 3. Crea tree
            _tree_payload: dict = {"tree": tree_items}
            if base_sha:
                _tree_payload["base_tree"] = base_sha
            _tree_r = await _c.post(f"{base_url}/git/trees", json=_tree_payload)
            if _tree_r.status_code not in (200, 201):
                return {"success": False, "error": f"Tree fail: {_tree_r.status_code}"}
            tree_sha = _tree_r.json()["sha"]

            # 4. Crea commit
            _commit_payload: dict = {"message": message, "tree": tree_sha}
            if base_sha:
                _commit_payload["parents"] = [base_sha]
            _commit_r = await _c.post(f"{base_url}/git/commits", json=_commit_payload)
            if _commit_r.status_code not in (200, 201):
                return {"success": False, "error": f"Commit fail: {_commit_r.status_code}"}
            commit_sha = _commit_r.json()["sha"]

            # 5. PATCH ref (o POST se branch nuovo)
            if branch_exists:
                _ref_upd = await _c.patch(f"{base_url}/git/refs/heads/{branch}", json={"sha": commit_sha})
            else:
                _ref_upd = await _c.post(f"{base_url}/git/refs", json={"ref": f"refs/heads/{branch}", "sha": commit_sha})
            if _ref_upd.status_code not in (200, 201):
                return {"success": False, "error": f"Ref update fail: {_ref_upd.status_code} — {_ref_upd.text[:200]}"}

            return {
                "success": True,
                "commit_sha": commit_sha,
                "branch": branch,
                "files_synced": len(tree_items),
                "repo": gh_repo,
                "url": f"https://github.com/{gh_repo}/tree/{branch}",
            }
    except Exception as _e:
        return {"success": False, "error": f"git_sync_vfs errore: {str(_e)[:300]}"}


async def _git_commit(message: str, cwd: str = ".", push: bool = False, add_all: bool = True) -> dict:
    """S763: git add -A + commit -m. push=True per git push. Risk medium."""
    try:
        async def _run(cmd: list) -> tuple:
            p = await asyncio.create_subprocess_exec(
                *cmd, cwd=cwd,
                stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE,
            )
            out, err = await asyncio.wait_for(p.communicate(), timeout=30)
            return p.returncode, (out + err).decode("utf-8", errors="replace")[:500]
        if add_all:
            rc, out = await _run(["git", "add", "-A"])
            if rc != 0:
                return {"ok": False, "step": "git add", "error": out}
        rc, out = await _run(["git", "commit", "-m", message])
        if rc != 0:
            return {"ok": False, "step": "git commit", "error": out}
        result: dict = {"ok": True, "commit_output": out}
        if push:
            rc_p, out_p = await _run(["git", "push"])
            result["push_ok"] = rc_p == 0
            result["push_output"] = out_p
        return result
    except Exception as e:
        return {"ok": False, "error": str(e)[:300]}


async def _npm_install(cwd: str = ".", manager: str = "auto", args: str = "") -> dict:
    """S763: npm/pnpm/yarn install. Auto-detecta da lockfile. Timeout 120s."""
    import os as _os
    try:
        if manager == "auto":
            manager = ("pnpm" if _os.path.exists(_os.path.join(cwd, "pnpm-lock.yaml"))
                       else "yarn" if _os.path.exists(_os.path.join(cwd, "yarn.lock"))
                       else "npm")
        cmd = [manager, "install"] + ([args] if args else [])
        proc = await asyncio.create_subprocess_exec(
            *cmd, cwd=cwd,
            stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE,
        )
        out, err = await asyncio.wait_for(proc.communicate(), timeout=120)
        combined = (out + err).decode("utf-8", errors="replace")[:2000]
        return {"ok": proc.returncode == 0, "manager": manager, "output": combined, "code": proc.returncode}
    except asyncio.TimeoutError:
        return {"ok": False, "error": f"{manager} install timeout (120s)"}
    except Exception as e:
        return {"ok": False, "error": str(e)[:300]}


async def _npm_run(script: str, cwd: str = ".", manager: str = "auto") -> dict:
    """S763: npm/pnpm/yarn run <script>. Auto-detecta manager. Timeout 120s."""
    import os as _os
    try:
        if manager == "auto":
            manager = ("pnpm" if _os.path.exists(_os.path.join(cwd, "pnpm-lock.yaml"))
                       else "yarn" if _os.path.exists(_os.path.join(cwd, "yarn.lock"))
                       else "npm")
        proc = await asyncio.create_subprocess_exec(
            manager, "run", script, cwd=cwd,
            stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE,
        )
        out, err = await asyncio.wait_for(proc.communicate(), timeout=120)
        combined = (out + err).decode("utf-8", errors="replace")[:3000]
        return {"ok": proc.returncode == 0, "script": script, "manager": manager,
                "output": combined, "code": proc.returncode}
    except asyncio.TimeoutError:
        return {"ok": False, "error": f"{manager} run {script} timeout (120s)"}
    except Exception as e:
        return {"ok": False, "error": str(e)[:300]}


async def _pip_install(packages: str, upgrade: bool = False, user: bool = False) -> dict:
    """S763: pip install. packages: stringa spazio/virgola separata. Timeout 120s."""
    try:
        cmd = [sys.executable, "-m", "pip", "install", "-q"]
        if upgrade:
            cmd.append("--upgrade")
        if user:
            cmd.append("--user")
        pkgs = [p.strip() for p in packages.replace(",", " ").split() if p.strip()]
        cmd.extend(pkgs)
        proc = await asyncio.create_subprocess_exec(
            *cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE,
        )
        out, err = await asyncio.wait_for(proc.communicate(), timeout=120)
        combined = (out + err).decode("utf-8", errors="replace")[:2000]
        return {"ok": proc.returncode == 0, "packages": pkgs, "output": combined, "code": proc.returncode}
    except asyncio.TimeoutError:
        return {"ok": False, "error": "pip install timeout (120s)"}
    except Exception as e:
        return {"ok": False, "error": str(e)[:300]}


async def _type_check(path: str = ".", checker: str = "auto", strict: bool = False) -> dict:
    """S763: tsc --noEmit o mypy. Auto-detecta da tsconfig.json/pyproject.toml. Timeout 60s."""
    import os as _os
    try:
        if checker == "auto":
            checker = ("tsc" if _os.path.exists(_os.path.join(path, "tsconfig.json")) else
                       "mypy" if (_os.path.exists(_os.path.join(path, "setup.py")) or
                                  _os.path.exists(_os.path.join(path, "pyproject.toml"))) else "tsc")
        if checker == "tsc":
            cmd = ["npx", "tsc", "--noEmit"] + (["--strict"] if strict else [])
        elif checker == "mypy":
            cmd = ["mypy", path, "--ignore-missing-imports"] + (["--strict"] if strict else [])
        else:
            return {"error": f"Checker non supportato: {checker}"}
        proc = await asyncio.create_subprocess_exec(
            *cmd, cwd=path,
            stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE,
        )
        out, err = await asyncio.wait_for(proc.communicate(), timeout=60)
        combined = (out + err).decode("utf-8", errors="replace")[:3000]
        ok = proc.returncode == 0
        return {"ok": ok, "checker": checker, "output": combined[:1000],
                "errors": combined if not ok else "", "code": proc.returncode}
    except asyncio.TimeoutError:
        return {"ok": False, "error": f"{checker} timeout (60s)"}
    except Exception as e:
        return {"ok": False, "error": str(e)[:300]}



# ─── GAP-B: scaffold_project ────────────────────────────────────────────
async def _scaffold_project(
    framework: str = "react",
    project_name: str = "my-project",
    target_dir: str = "/tmp",
) -> dict:
    """GAP-B: Scaffolding template-based senza token LLM.

    Genera file boilerplate per framework comuni:
    react | nextjs | fastapi | flask | django | express | astro | sveltekit
    """
    import os as _os, re as _re_scaf

    _fw   = framework.lower().strip()
    _safe = _re_scaf.sub(r"[^a-z0-9\-]", "-", project_name.lower())[:30] or "my-project"
    _base = _os.path.join(target_dir, _safe)
    _os.makedirs(_base, exist_ok=True)
    _pn = project_name

    _TEMPLATES: dict[str, dict[str, str]] = {
        "react": {
            "package.json": (
                '{"name":"' + _safe + '","version":"0.1.0","type":"module",'
                '"scripts":{"dev":"vite","build":"vite build"},'
                '"dependencies":{"react":"^18.3.1","react-dom":"^18.3.1"},'
                '"devDependencies":{"@vitejs/plugin-react":"^4.3.1","typescript":"^5.5.0","vite":"^5.4.0"}}'
            ),
            "index.html": (
                '<!DOCTYPE html>\n<html lang="it">\n<head><meta charset="UTF-8">'
                '<title>' + _pn + '</title></head>\n'
                '<body><div id="root"></div>'
                '<script type="module" src="/src/main.tsx"></script></body>\n</html>'
            ),
            "src/main.tsx": (
                'import { StrictMode } from "react";\n'
                'import { createRoot } from "react-dom/client";\n'
                'import App from "./App";\n'
                'createRoot(document.getElementById("root")!).render(<StrictMode><App /></StrictMode>);'
            ),
            "src/App.tsx": (
                'export default function App() {\n'
                '  return <div style={{textAlign:"center",padding:"2rem"}}>\n'
                '    <h1>' + _pn + '</h1>\n'
                '    <p>Modifica <code>src/App.tsx</code> per iniziare.</p>\n'
                '  </div>;\n}'
            ),
            "src/index.css": "body{margin:0;font-family:system-ui,sans-serif}",
            "vite.config.ts": (
                'import { defineConfig } from "vite";\nimport react from "@vitejs/plugin-react";\n'
                'export default defineConfig({ plugins: [react()] });'
            ),
        },
        "nextjs": {
            "package.json": (
                '{"name":"' + _safe + '","version":"0.1.0",'
                '"scripts":{"dev":"next dev","build":"next build","start":"next start"},'
                '"dependencies":{"next":"^15.1.0","react":"^19","react-dom":"^19"},'
                '"devDependencies":{"typescript":"^5","@types/node":"^20","@types/react":"^19",\"@types/react-dom\":\"^19\"}}'
            ),
            "app/layout.tsx": (
                'export const metadata = { title: "' + _pn + '" };\n'
                'export default function Layout({ children }: { children: React.ReactNode }) {\n'
                '  return <html lang="it"><body>{children}</body></html>;\n}'
            ),
            "app/page.tsx": (
                '"use client";\n'
                'export default function Page() {\n'
                '  return <main style={{padding:"2rem"}}><h1>' + _pn + '</h1></main>;\n}'
            ),
            "next.config.mjs": "const nextConfig = {};\nexport default nextConfig;",
        },
        "fastapi": {
            "main.py": (
                'from fastapi import FastAPI\nfrom fastapi.middleware.cors import CORSMiddleware\n\n'
                'app = FastAPI(title="' + _pn + '", version="0.1.0")\n'
                'app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"])\n\n'
                '@app.get("/health")\nasync def health(): return {"status": "ok"}\n\n'
                '@app.get("/")\nasync def root(): return {"message": "Benvenuto in ' + _pn + '"}\n'
            ),
            "requirements.txt": "fastapi>=0.115.0\nuvicorn[standard]>=0.30.0\nhttpx>=0.27.0\n",
            "Dockerfile": (
                'FROM python:3.11-slim\nWORKDIR /app\n'
                'COPY requirements.txt .\nRUN pip install -r requirements.txt\n'
                'COPY . .\nEXPOSE 8000\nCMD ["uvicorn","main:app","--host","0.0.0.0","--port","8000"]'
            ),
            ".gitignore": "__pycache__/\n*.pyc\n.env\n",
        },
        "flask": {
            "app.py": (
                'from flask import Flask, jsonify\nfrom flask_cors import CORS\n\n'
                'app = Flask(__name__)\nCORS(app)\n\n'
                '@app.get("/health")\ndef health(): return jsonify({"status": "ok"})\n\n'
                '@app.get("/")\ndef root(): return jsonify({"message": "Benvenuto in ' + _pn + '"})\n\n'
                'if __name__ == "__main__":\n    app.run(debug=True, host="0.0.0.0", port=5000)\n'
            ),
            "requirements.txt": "flask>=3.0.0\nflask-cors>=4.0.0\ngunicorn>=21.2.0\n",
            ".gitignore": "__pycache__/\n*.pyc\n.env\n",
        },
        "django": {
            "manage.py": (
                '#!/usr/bin/env python\nimport os, sys\n'
                'os.environ.setdefault("DJANGO_SETTINGS_MODULE","config.settings")\n'
                'from django.core.management import execute_from_command_line\nexecute_from_command_line(sys.argv)\n'
            ),
            "requirements.txt": "django>=5.0\ndjangorestframework>=3.15\ndjango-cors-headers>=4.3\ngunicorn>=21.2.0\n",
            "config/__init__.py": "",
            "config/settings.py": (
                'from pathlib import Path\nBASE_DIR=Path(__file__).resolve().parent.parent\n'
                'SECRET_KEY="change-me-in-production"\nDEBUG=True\nALLOWED_HOSTS=["*"]\n'
                'INSTALLED_APPS=["django.contrib.contenttypes","django.contrib.auth","rest_framework","corsheaders"]\n'
                'MIDDLEWARE=["corsheaders.middleware.CorsMiddleware","django.middleware.common.CommonMiddleware"]\n'
                'ROOT_URLCONF="config.urls"\nDEFAULT_AUTO_FIELD="django.db.models.BigAutoField"\nCORS_ALLOW_ALL_ORIGINS=True\n'
            ),
            "config/urls.py": 'from django.urls import path, include\nurlpatterns=[path("api/", include("api.urls"))]\n',
            "api/__init__.py": "",
            "api/views.py": (
                'from rest_framework.decorators import api_view\nfrom rest_framework.response import Response\n\n'
                '@api_view(["GET"])\n'
                'def hello(request): return Response({"message": "Benvenuto in ' + _pn + '"})\n'
            ),
            "api/urls.py": 'from django.urls import path\nfrom . import views\nurlpatterns=[path("", views.hello)]\n',
        },
        "express": {
            "package.json": (
                '{"name":"' + _safe + '","version":"0.1.0","type":"module",'
                '"scripts":{"start":"node src/index.js","dev":"node --watch src/index.js"},'
                '"dependencies":{"express":"^4.19.2"}}'
            ),
            "src/index.js": (
                'import express from "express";\n'
                'const app=express(), PORT=process.env.PORT||3000;\n'
                'app.use(express.json());\n'
                'app.get("/health", (_, res) => res.json({ status: "ok" }));\n'
                'app.get("/", (_, res) => res.json({ message: "Benvenuto in ' + _pn + '" }));\n'
                'app.listen(PORT, () => console.log("Server: http://localhost:" + PORT));\n'
            ),
            ".gitignore": "node_modules/\n.env\n",
        },
        "astro": {
            "package.json": (
                '{"name":"' + _safe + '","version":"0.1.0","type":"module",'
                '"scripts":{"dev":"astro dev","build":"astro build","preview":"astro preview"},'
                '"dependencies":{"astro":"^4.11.0"}}'
            ),
            "astro.config.mjs": (
                'import { defineConfig } from "astro/config";\n'
                'export default defineConfig({});\n'
            ),
            "src/pages/index.astro": (
                '---\nconst title = "' + _pn + '";\n---\n'
                '<html lang="it">\n  <head><meta charset="UTF-8"><title>{title}</title></head>\n'
                '  <body>\n    <h1>{title}</h1>\n'
                '    <p>Modifica <code>src/pages/index.astro</code> per iniziare.</p>\n'
                '  </body>\n</html>\n'
            ),
            "src/layouts/Layout.astro": (
                '---\nconst { title } = Astro.props;\n---\n'
                '<!DOCTYPE html>\n<html lang="it">\n'
                '  <head><meta charset="UTF-8"><title>{title}</title></head>\n'
                '  <body><slot /></body>\n</html>\n'
            ),
            ".gitignore": "node_modules/\ndist/\n.astro/\n.env\n",
        },
        "sveltekit": {
            "package.json": (
                '{"name":"' + _safe + '","version":"0.1.0","type":"module",'
                '"scripts":{"dev":"vite dev","build":"vite build","preview":"vite preview"},'
                '"dependencies":{"@sveltejs/kit":"^2.5.0","svelte":"^4.2.0"},'
                '"devDependencies":{"@sveltejs/adapter-auto":"^3.2.0","vite":"^5.3.0"}}'
            ),
            "svelte.config.js": (
                'import adapter from "@sveltejs/adapter-auto";\n'
                'export default { kit: { adapter: adapter() } };\n'
            ),
            "vite.config.js": (
                'import { sveltekit } from "@sveltejs/kit/vite";\n'
                'import { defineConfig } from "vite";\n'
                'export default defineConfig({ plugins: [sveltekit()] });\n'
            ),
            "src/routes/+page.svelte": (
                '<script>\n  const title = "' + _pn + '";\n</script>\n'
                '<main>\n  <h1>{title}</h1>\n'
                '  <p>Modifica <code>src/routes/+page.svelte</code> per iniziare.</p>\n'
                '</main>\n'
            ),
            "src/routes/+layout.svelte": '<slot />\n',
            ".gitignore": "node_modules/\nbuild/\n.svelte-kit/\n.env\n",
        },
    }

    _match = 'react'
    for _k in _TEMPLATES:
        if _k in _fw or _fw.startswith(_k[:4]):
            _match = _k
            break

    _tpl = _TEMPLATES[_match]
    _created: list[str] = []
    _errs_scaf: list[str] = []
    for _rel, _content in _tpl.items():
        _full = _os.path.join(_base, _rel)
        _os.makedirs(_os.path.dirname(_full), exist_ok=True)
        try:
            with open(_full, 'w', encoding='utf-8') as _f:
                _f.write(_content)
            _created.append(_rel)
        except Exception as _e:
            _errs_scaf.append(f'{_rel}: {_e}')

    _steps_map = {
        'react':   'npm install && npm run dev',
        'nextjs':  'npm install && npm run dev',
        'fastapi': 'pip install -r requirements.txt && uvicorn main:app --reload',
        'flask':   'pip install -r requirements.txt && python app.py',
        'django':  'pip install -r requirements.txt && python manage.py runserver',
        'express':   'npm install && npm run dev',
        'astro':     'npm install && npm run dev',
        'sveltekit': 'npm install && npm run dev',
    }
    _next_step = _steps_map.get(_match, 'installa le dipendenze')
    _out = (
        f'Scaffold **{_match}** per **{_pn}** — {len(_created)} file creati:\n'
        + '\n'.join(f'  {p}' for p in _created)
        + f'\n\nDirectory: {_base}'
        + f'\n\nProssimi passi: cd {_safe} && {_next_step}'
    )
    if _errs_scaf:
        _out += f'\n\nErrori: {"; ".join(_errs_scaf)}'
    # GAP-X6: restituisce anche il dict files → usato da /api/scaffold_project
    # Il frontend li scrive nel VFS con vfsAsync.write() — zero passaggi via agent.
    return {
        'success':      True,
        'output':       _out,
        'files':        _tpl,          # dict {rel_path: content} — già con project_name interpolato
        'framework':    _match,
        'project_name': _pn,
        'project_dir':  f'/{_safe}',   # percorso VFS suggerito
        'created':      _created,
    }

# ─── S-GAP15: create_chart ──────────────────────────────────────────────────
async def _create_chart(
    chart_type: str = "bar",
    data: dict | None = None,
    title: str = "",
    x_label: str = "",
    y_label: str = "",
    labels: list | None = None,
    values: list | None = None,
) -> dict:
    """
    S-GAP15: Genera un grafico come immagine PNG base64 usando matplotlib (se disponibile),
    fallback SVG testuale per ambienti senza display.
    chart_type: bar | line | pie | scatter
    data: dict {label: value} oppure usa labels/values separati
    """
    import base64
    import io

    # Normalizza input
    if data and isinstance(data, dict):
        _labels = list(data.keys())
        _values = [float(v) for v in data.values()]
    else:
        _labels = labels or []
        _values = [float(v) for v in (values or [])]

    if not _labels or not _values:
        return {"error": "Devi fornire 'data' (dict) oppure 'labels' e 'values' (list)."}

    # Tentativo matplotlib (potrebbe non essere disponibile in tutti gli ambienti)
    try:
        import matplotlib
        matplotlib.use("Agg")  # Headless — nessun display necessario
        import matplotlib.pyplot as plt

        fig, ax = plt.subplots(figsize=(8, 5))
        ct = chart_type.lower().strip()
        if ct == "bar":
            ax.bar(_labels, _values)
        elif ct == "line":
            ax.plot(_labels, _values, marker="o")
        elif ct == "pie":
            ax.pie(_values, labels=_labels, autopct="%1.1f%%")
        elif ct == "scatter":
            ax.scatter(range(len(_values)), _values)
            ax.set_xticks(range(len(_labels)))
            ax.set_xticklabels(_labels, rotation=45, ha="right")
        else:
            ax.bar(_labels, _values)  # default bar

        if title:   ax.set_title(title)
        if x_label: ax.set_xlabel(x_label)
        if y_label: ax.set_ylabel(y_label)
        plt.tight_layout()

        buf = io.BytesIO()
        fig.savefig(buf, format="png", dpi=100)
        plt.close(fig)
        buf.seek(0)
        b64 = base64.b64encode(buf.read()).decode("utf-8")
        return {
            "type": "image/png",
            "format": "base64",
            "data": b64,
            "chart_type": ct,
            "title": title,
            "note": f"Grafico {ct} generato con matplotlib. Mostra l'immagine con: <img src='data:image/png;base64,{b64[:30]}...'>",
        }
    except ImportError:
        pass  # matplotlib non disponibile → fallback SVG

    # Fallback SVG testuale (sempre disponibile)
    _max_v = max(_values) if _values else 1
    _bar_w = 60
    _gap   = 10
    _h     = 200
    _svg_w = len(_labels) * (_bar_w + _gap) + 60
    bars = ""
    for i, (lbl, val) in enumerate(zip(_labels, _values)):
        bh   = int((_h - 40) * val / _max_v) if _max_v else 1
        x    = 40 + i * (_bar_w + _gap)
        y    = _h - bh - 20
        bars += f'<rect x="{x}" y="{y}" width="{_bar_w}" height="{bh}" fill="#4f8ef7"/>'
        bars += f'<text x="{x + _bar_w//2}" y="{_h - 5}" text-anchor="middle" font-size="11">{str(lbl)[:10]}</text>'
        bars += f'<text x="{x + _bar_w//2}" y="{y - 4}" text-anchor="middle" font-size="10">{val}</text>'
    title_tag = f'<text x="{_svg_w//2}" y="18" text-anchor="middle" font-size="14" font-weight="bold">{title}</text>' if title else ""
    svg = f'<svg xmlns="http://www.w3.org/2000/svg" width="{_svg_w}" height="{_h + 30}" style="background:#fff">{title_tag}{bars}</svg>'
    import base64 as _b64
    svg_b64 = _b64.b64encode(svg.encode()).decode()
    return {
        "type": "image/svg+xml",
        "format": "base64",
        "data": svg_b64,
        "chart_type": "bar_svg_fallback",
        "title": title,
        "note": "matplotlib non disponibile — grafico SVG testuale (fallback). Installa matplotlib per grafici PNG di qualità.",
    }



# ─── P17-F4-REG: trigger_webhook — registra il tool nel registry ────────────

# ─── P24-F1: Macro tools — workflow compositi ─────────────────────────────────

async def _write_and_check(
    path: str,
    content: str,
    language: str = "auto",
    run_lint: bool = True,
) -> dict:
    """Macro: write_file → lint_code — scrive e verifica in un unico step."""
    write_res = await _write_file(path=path, content=content)
    lint_res: "dict | None" = None
    if run_lint:
        ext = path.rsplit(".", 1)[-1].lower() if "." in path else ""
        if ext in ("py", "js", "ts", "tsx", "jsx", "json", "css", "html"):
            lint_res = await _lint_code_tool(content=content, language=language, path=path)
    has_errors = bool(lint_res and (lint_res.get("errors") or lint_res.get("error")))
    return {
        "ok": True,
        "path": path,
        "written": True,
        "bytes_written": len(content.encode("utf-8")),
        "lint": lint_res,
        "has_errors": has_errors,
        "message": (
            f"Scritto {path} ({len(content)} chars)"
            + (" — lint OK" if lint_res and not has_errors else " — errori lint" if has_errors else "")
        ),
    }


async def _bulk_write_files(
    files: dict,
    stop_on_error: bool = False,
    run_lint: bool = False,
) -> dict:
    """Macro: write_file x N in parallelo — scrive più file in un solo step."""
    if not isinstance(files, dict) or not files:
        return {"ok": False, "error": "files deve essere un dict {path: content} non vuoto"}

    async def _single(p: str, c: str) -> "tuple[str, bool, str]": # type: ignore[type-arg]
        try:
            await _write_file(path=p, content=str(c))
            if run_lint:
                ext = p.rsplit(".", 1)[-1].lower() if "." in p else ""
                if ext in ("py", "js", "ts", "tsx", "jsx", "json"):
                    lr = await _lint_code_tool(content=str(c), path=p)
                    if lr.get("errors"):
                        return p, False, f"lint errors: {str(lr['errors'])[:200]}"
            return p, True, ""
        except Exception as exc:  # noqa: BLE001
            return p, False, str(exc)[:200]

    results = await asyncio.gather(*[_single(p, c) for p, c in files.items()])
    written = [p for p, ok, _ in results if ok]
    failed  = [{"path": p, "error": e} for p, ok, e in results if not ok]
    return {
        "ok": len(failed) == 0,
        "written": written,
        "failed": failed,
        "total": len(files),
        "written_count": len(written),
        "message": f"Scritti {len(written)}/{len(files)} file" + (f" — {len(failed)} errori" if failed else ""),
    }


async def _fetch_and_extract(url: str, extract: str = "all") -> dict:
    """Macro: read_page → estrazione strutturata (title, testo, word_count, links)."""
    import re as _re
    page = await _read_page(url=url, query="")
    raw_text = page.get("text") or page.get("content") or ""
    title    = page.get("title", "")
    paras    = [p.strip() for p in raw_text.split("\n") if len(p.strip()) > 60]
    result: dict = {
        "ok": page.get("ok", True),
        "url": url,
        "title": title,
        "word_count": len(raw_text.split()),
        "char_count": len(raw_text),
    }
    if page.get("error"):
        result["error"] = page["error"]
    if extract in ("all", "text", "summary"):
        result["paragraphs"] = paras[:8]
        result["excerpt"]    = raw_text[:2500]
    if extract in ("all", "links"):
        result["links"] = list(dict.fromkeys(_re.findall(r"https?://[^\s'\"<>]{10,}", raw_text)))[:30]
    return result


async def _read_multiple_files(paths: list, max_chars_per_file: int = 4000) -> dict:
    """Macro: read_file x N in parallelo — legge più file e aggrega i contenuti."""
    if not paths:
        return {"ok": False, "error": "paths non può essere vuoto"}

    async def _single(path: str) -> "tuple[str, str | None, str | None]": # type: ignore[type-arg]
        try:
            res = await _read_file(path=path)
            content = (res.get("content") or res.get("text") or "")[:max_chars_per_file]
            return path, content, None
        except Exception as exc:  # noqa: BLE001
            return path, None, str(exc)[:200]

    results = await asyncio.gather(*[_single(p) for p in paths])
    files_out  = {p: c for p, c, e in results if c is not None}
    errors_out = {p: e for p, c, e in results if e is not None}
    return {
        "ok": len(errors_out) == 0,
        "files": files_out,
        "errors": errors_out,
        "read_count": len(files_out),
        "total": len(paths),
    }


async def _trigger_webhook(
    url: str,
    payload: "dict | str | None" = None,
    method: str = "POST",
    headers: "dict | None" = None,
    timeout: float = 10.0,
) -> dict:
    """Wrapper — delega all'implementazione in tools/trigger_webhook.py."""
    try:
        from tools.trigger_webhook import trigger_webhook as _tw
        return await _tw(url=url, payload=payload, method=method, headers=headers, timeout=timeout)
    except ImportError:
        import httpx as _hx
        body = None
        _hdrs: dict = {"User-Agent": "agente-ai/1.0"}
        if headers:
            _hdrs.update(headers)
        if payload is not None:
            import json as _j
            body = _j.dumps(payload).encode() if isinstance(payload, dict) else str(payload).encode()
            _hdrs.setdefault("Content-Type", "application/json")
        async with _hx.AsyncClient(timeout=min(float(timeout), 10.0), follow_redirects=True) as _c:
            _m = method.upper()
            if _m == "GET":
                r = await _c.get(url, headers=_hdrs)
            elif _m == "PUT":
                r = await _c.put(url, content=body, headers=_hdrs)
            else:
                r = await _c.post(url, content=body, headers=_hdrs)
        return {"ok": r.is_success, "status": r.status_code, "body": r.text[:2000], "error": None}


# ─── P24-F2: notion_rw wrapper ────────────────────────────────────────────────
async def _notion_rw(
    action: str,
    query: str = "",
    page_id: str = "",
    parent_id: str = "",
    title: str = "",
    content: str = "",
    max_results: int = 5,
) -> dict:
    """Wrapper — delega all'implementazione in tools/notion_tool.py."""
    from tools.notion_tool import notion_rw as _nrw  # noqa: PLC0415
    return await _nrw(
        action=action, query=query, page_id=page_id,
        parent_id=parent_id, title=title, content=content,
        max_results=max_results,
    )


# ─── P24-F3: jina_fetch wrapper ────────────────────────────────────────────────
async def _jina_fetch(
    url: str,
    query: str           = "",
    target_selector: str = "",
    remove_selector: str = "",
    max_length: int      = 12000,
) -> dict:
    """Wrapper — delega a tools/jina_reader.py (Jina Reader per SPA e siti JS)."""
    from tools.jina_reader import jina_fetch as _jr  # noqa: PLC0415
    return await _jr(
        url=url, query=query,
        target_selector=target_selector,
        remove_selector=remove_selector,
        max_length=max_length,
    )

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:
    - Syntax check con posizione esatta (riga, colonna, testo)
    - Metriche strutturali: funzioni/classi/imports/nesting/righe
    - 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 (parametro 'code' o 'content' richiesto)", "line": 0}]
        result["summary"] = "Codice vuoto"
        return result

    # 1. Syntax check
    try:
        tree = _ast.parse(code, filename=filename)
        result["syntax_ok"] = True
    except SyntaxError as _e:
        result["errors"] = [{
            "type": "SyntaxError", "message": str(_e.msg or _e),
            "line": _e.lineno or 0, "col": _e.offset or 0,
            "text": (_e.text or "").rstrip(),
        }]
        result["summary"] = f"SyntaxError alla riga {_e.lineno}: {_e.msg}"
        return result
    except IndentationError as _e:
        result["errors"] = [{
            "type": "IndentationError", "message": str(_e.msg or _e), "line": _e.lineno or 0,
        }]
        result["summary"] = f"IndentationError alla riga {_e.lineno}"
        return result

    # 2. AST visitor per metriche
    class _V(_ast.NodeVisitor):
        def __init__(self):
            self.fns: list[dict] = []
            self.classes: list[dict] = []
            self.imports: list[str] = []
            self.max_nesting = 0
            self._depth = 0
        def _ent(self): self._depth += 1; self.max_nesting = max(self.max_nesting, self._depth)
        def _ex(self):  self._depth -= 1
        def visit_FunctionDef(self, n):
            _ln = (n.end_lineno or n.lineno) - n.lineno + 1
            self.fns.append({"name": n.name, "line": n.lineno, "lines": _ln})
            self._ent(); self.generic_visit(n); self._ex()
        visit_AsyncFunctionDef = visit_FunctionDef
        def visit_ClassDef(self, n):
            self.classes.append({"name": n.name, "line": n.lineno})
            self._ent(); self.generic_visit(n); self._ex()
        def visit_For(self, n):   self._ent(); self.generic_visit(n); self._ex()
        def visit_While(self, n): self._ent(); self.generic_visit(n); self._ex()
        def visit_If(self, n):    self._ent(); self.generic_visit(n); self._ex()
        def visit_With(self, n):  self._ent(); self.generic_visit(n); self._ex()
        def visit_Try(self, n):   self._ent(); self.generic_visit(n); self._ex()
        def visit_Import(self, n):
            for _a in n.names: self.imports.append(_a.name)
        def visit_ImportFrom(self, n):
            if n.module: self.imports.append(n.module)

    _v = _V(); _v.visit(tree)
    _tot = len(code.splitlines())
    result["complexity"] = {
        "total_lines": _tot,
        "functions": len(_v.fns),
        "classes": len(_v.classes),
        "imports": _v.imports[:20],
        "max_nesting": _v.max_nesting,
    }

    # 3. Suggerimenti actionable
    _sug: list[str] = []
    for _f in _v.fns:
        if _f["lines"] > 50:
            _sug.append(f"Funzione '{_f['name']}' (riga {_f['line']}) ha {_f['lines']} righe — valuta di dividerla")
    if _v.max_nesting >= 5:
        _sug.append(f"Nesting max {_v.max_nesting} livelli — rischio complessità ciclomatica alta")
    if not _v.fns and _tot > 30:
        _sug.append("Nessuna funzione definita su codice lungo — struttura in funzioni")
    if "import *" in code:
        _sug.append("Evita 'import *' — importa esplicitamente solo ciò che serve")
    _dup_imports = {_i for _i in _v.imports if _v.imports.count(_i) > 1}
    if _dup_imports:
        _sug.append(f"Import duplicati rilevati: {', '.join(sorted(_dup_imports))}")
    result["suggestions"] = _sug[:5]

    # 4. Summary
    _parts = [f"OK — {_tot} righe"]
    if _v.fns:    _parts.append(f"{len(_v.fns)} funzioni")
    if _v.classes:_parts.append(f"{len(_v.classes)} classi")
    if _v.imports:_parts.append(f"{len(_v.imports)} import")
    if _v.max_nesting: _parts.append(f"nesting max {_v.max_nesting}")
    result["summary"] = "Sintassi " + ", ".join(_parts)
    return result


TOOL_REGISTRY: dict[str, dict] = {
    "web_search": {
        "name": "web_search",
        "goal": "Cerca informazioni aggiornate sul web",
        "description": "Ricerca web multi-provider: Brave Search → Tavily → DuckDuckGo → SearXNG. Brave/Tavily attivi se BRAVE_SEARCH_API_KEY/TAVILY_API_KEY sono impostati come env secrets HF Spaces.",
        "required_inputs": ["query"],
        "optional_inputs": {"max_results": 5},
        "risk_level": "low",
        "fallbacks": ["direct_response"],
        "_fn": _web_search,
    },
    "read_page": {
        "name": "read_page",
        "goal": "Legge il contenuto di una pagina web",
        "description": "Scarica e pulisce il testo di una URL",
        "required_inputs": ["url"],
        "optional_inputs": {},
        "risk_level": "low",
        "fallbacks": ["web_search"],
        "_fn": _read_page,
    },
    "get_weather": {
        "name": "get_weather",
        "goal": "Ottieni meteo attuale per una città",
        "description": "Usa Open-Meteo (gratuito, no API key) per meteo in tempo reale",
        "required_inputs": ["city"],
        "optional_inputs": {},
        "risk_level": "low",
        "fallbacks": [],
        "_fn": _get_weather,
    },
    "calculate": {
        "name": "calculate",
        "goal": "Calcola espressioni matematiche in modo sicuro",
        "description": "Valuta espressioni matematiche senza exec() — solo operatori sicuri",
        "required_inputs": ["expression"],
        "optional_inputs": {},
        "risk_level": "low",
        "fallbacks": [],
        "_fn": _calculate,
    },
    "run_python": {
        "name": "run_python",
        "goal": "Esegui codice Python reale sul server",
        "description": "Esegue Python vero: calcoli, data processing, file I/O, librerie stdlib. Timeout 15s.",
        "required_inputs": ["code"],
        "optional_inputs": {},
        "risk_level": "medium",
        "fallbacks": [],
        "_fn": _run_python,
    },
    "generate_image": {
        "name": "generate_image",
        "goal": "Genera un'immagine AI a partire da una descrizione testuale",
        "description": "Usa Pollinations AI (gratuito, no API key). Restituisce URL immagine PNG pronto.",
        "required_inputs": ["prompt"],
        "optional_inputs": {"width": 512, "height": 512},
        "risk_level": "low",
        "fallbacks": [],
        "_fn": _generate_image,
    },
    "get_news": {
        "name": "get_news",
        "goal": "Recupera notizie recenti su un argomento",
        "description": "S378: alias di web_search ottimizzato per notizie. Aggiunge 'notizie recenti' alla query se non già presente.",
        "required_inputs": ["query"],
        "optional_inputs": {"max_results": 5},
        "risk_level": "low",
        "fallbacks": ["web_search"],
        "_fn": _get_news,
    },

    # ── S403: Browser tools — Playwright esposto come tool ──────────────────────
    # Limite 1 (criticità 10/10) e Limite 4 (8/10) del documento S403.
    # Playwright era già in backend/api/browser.py ma non era mai un tool dell'agente.
    # Decisione architetturale: API-first (web_search/read_page) → browser se API insufficiente.

    "browser_navigate": {
        "name": "browser_navigate",
        "goal": "Apre una pagina web reale con browser headless e ne legge il contenuto",
        "description": (
            "S403: Playwright headless — naviga a URL, restituisce titolo, testo principale "
            "(max 2000 chars), lista link (max 15) e input interattivi (max 10). "
            "Stateless (no sessione). Usa quando read_page è insufficiente per siti dinamici/SPA. "
            "API-first: preferisci web_search/read_page se il sito ha API pubbliche."
        ),
        "required_inputs": ["url"],
        "optional_inputs": {"wait_ms": 2000, "mobile": False},
        "risk_level": "medium",
        "fallbacks": ["read_page", "web_search"],
        "_fn": _browser_navigate,
    },
    "browser_session_open": {
        "name": "browser_session_open",
        "goal": "Apre sessione browser persistente per task multi-step (login, form, checkout)",
        "description": (
            "S403: Playwright sessione stateful. Restituisce session_id da usare con "
            "browser_session_act per ogni step successivo (click/fill/submit). "
            "Usa browser_session_close quando hai finito. Max 2 sessioni attive. "
            "TTL automatico: 8 minuti di inattività → chiusura."
        ),
        "required_inputs": ["url"],
        "optional_inputs": {"wait_ms": 1500, "mobile": False},
        "risk_level": "high",
        "fallbacks": ["browser_navigate"],
        "_fn": _browser_session_open,
    },
    "browser_session_act": {
        "name": "browser_session_act",
        "goal": "Esegue azioni (click/fill/scroll) su sessione browser aperta",
        "description": (
            "S403: Esegue azioni su sessione Playwright aperta con browser_session_open. "
            "actions: lista [{type, selector, value, key, ms}]. "
            "Tipi: click, fill, select, press, hover, wait_for, wait, scroll. "
            "Restituisce DOM aggiornato dopo le azioni."
        ),
        "required_inputs": ["session_id", "actions"],
        "optional_inputs": {"wait_ms": 1000},
        "risk_level": "high",
        "fallbacks": [],
        "_fn": _browser_session_act,
    },

    "web_research": {
        "name": "web_research",
        "goal": "Ricerca approfondita multi-fonte su un argomento con sintesi AI",
        "description": (
            "Cerca N fonti web su un topic, ne estrae il contenuto rilevante e produce "
            "una sintesi AI (Groq). Più approfondito di web_search + read_page. "
            "Usa per ricerche che richiedono confronto tra più fonti."
        ),
        "required_inputs": ["topic"],
        "optional_inputs": {"depth": 4, "synthesize": True},
        "risk_level": "low",
        "fallbacks": ["web_search"],
        "_fn": _web_research,
    },
    "send_email": {
        "name": "send_email",
        "goal": "Invia email transazionale via Resend API",
        "description": (
            "Invia email a un destinatario via Resend. "
            "Richiede RESEND_API_KEY nell'ambiente HF Space. "
            "Supporta testo plain e HTML."
        ),
        "required_inputs": ["to", "subject", "body"],
        "optional_inputs": {"html": False, "from_name": "Agente AI"},
        "risk_level": "medium",
        "fallbacks": [],
        "_fn": _send_email,
    },
    "database_query": {
        "name": "database_query",
        "goal": "Esegui query SQL su database configurato (PostgreSQL/SQLite)",
        "description": (
            "Esegue query SQL su DATABASE_URL configurato nell'ambiente. "
            "Default read_only=True (solo SELECT). "
            "Richiede DATABASE_URL env var (postgresql:// o sqlite:///path)."
        ),
        "required_inputs": ["sql"],
        "optional_inputs": {"params": [], "read_only": True},
        "risk_level": "medium",
        "fallbacks": [],
        "_fn": _database_query,
    },

    "call_api": {
        "name": "call_api",
        "goal": "Chiama qualsiasi API REST esterna (GET/POST/PUT/PATCH/DELETE)",
        "description": (
            "S601: Chiama endpoint REST con metodo/headers/body/autenticazione configurabili. "
            "Supporta auth bearer, basic, api_key. Restituisce status, headers, body JSON/testo. "
            "Usa per integrare webhook, API pubbliche, testare endpoint o inviare dati."
        ),
        "required_inputs": ["url"],
        "optional_inputs": {"method": "GET", "headers": {}, "body": None,
                             "auth_type": "none", "auth_value": "", "timeout_ms": 15000},
        "risk_level": "medium",
        "fallbacks": ["web_search"],
        "_fn": _call_api,
    },
    "execute_sql": {
        "name": "execute_sql",
        "goal": "Esegui query SQL su database (alias di database_query)",
        "description": (
            "S601: Esegue SQL su DATABASE_URL configurato (PostgreSQL/SQLite). "
            "Default read_only=True — solo SELECT. Alias semantico di database_query."
        ),
        "required_inputs": ["sql"],
        "optional_inputs": {"params": [], "read_only": True},
        "risk_level": "medium",
        "fallbacks": ["database_query"],
        "_fn": _execute_sql,
    },
    "create_pdf": {
        "name": "create_pdf",
        "goal": "Genera un PDF da HTML, Markdown o testo",
        "description": (
            "S601: Crea PDF da contenuto HTML/Markdown/testo via backend. "
            "Restituisce URL o base64 del PDF generato. "
            "Usa per report, documenti, contratti, presentazioni."
        ),
        "required_inputs": ["content"],
        "optional_inputs": {"filename": "documento.pdf", "format": "html"},
        "risk_level": "low",
        "fallbacks": [],
        "_fn": _create_pdf,
    },
    # S666: tool file-system — assenti in TOOL_REGISTRY ma presenti in unified_loop._TOOL_MAP.
    # Senza entries qui, agent.py riceve KeyError su lookup → tool ignorato silenziosamente.
    "read_file": {
        "name": "read_file",
        "goal": "Legge il contenuto di un file dal filesystem del backend",
        "description": "Legge un file locale del backend. Limit 10MB. Restituisce content+size.",
        "required_inputs": ["path"],
        "optional_inputs": {"encoding": "utf-8"},
        "risk_level": "low",
        "fallbacks": [],
        "_fn": _read_file,
    },
    "write_file": {
        "name": "write_file",
        "goal": "Scrive o sovrascrive un file nel filesystem del backend",
        "description": "Scrive testo in un file locale. Crea directory intermedie automaticamente.",
        "required_inputs": ["path", "content"],
        "optional_inputs": {"encoding": "utf-8"},
        "risk_level": "medium",
        "fallbacks": [],
        "_fn": _write_file,
    },
    "apply_patch": {
        "name": "apply_patch",
        "goal": "Applica una patch unified-diff a un file esistente",
        "description": (
            "Applica patch con subprocess patch(1) con fallback Python replace-based. "
            "Input: path (file da patchare) + patch (testo unified-diff). "
            "Risk medium: modifica file in modo difficilmente reversibile."
        ),
        "required_inputs": ["path", "patch"],
        "optional_inputs": {},
        "risk_level": "medium",
        "fallbacks": ["write_file"],
        "_fn": _apply_patch,
    },
    "execute_shell": {
        "name": "execute_shell",
        "goal": "Esegue un comando shell sul server backend",
        "description": (
            "Esegue comando arbitrario con subprocess. "
            "Timeout max 120s. Restituisce stdout/stderr/code. "
            "Risk HIGH: esecuzione arbitraria sul server."
        ),
        "required_inputs": ["command"],
        "optional_inputs": {"timeout": 30, "cwd": "."},
        "risk_level": "high",
        "fallbacks": [],
        "_fn": _execute_shell,
    },
    # ─── S763: 10 tool mancanti aggiunti ────────────────────────────────────
    "directory_tree": {
        "name": "directory_tree",
        "goal": "Visualizza struttura cartelle del progetto (albero file)",
        "description": "S763: os.walk — albero ASCII, max_depth 3. Ignora .git, node_modules, __pycache__.",
        "required_inputs": [],
        "optional_inputs": {"path": ".", "max_depth": 3, "show_hidden": False},
        "risk_level": "low",
        "fallbacks": ["file_search"],
        "_fn": _directory_tree,
    },
    "file_search": {
        "name": "file_search",
        "goal": "Cerca testo/pattern nei file del progetto (grep)",
        "description": "S763: grep -rn con fallback Python. pattern: stringa o regex. Max 50 match.",
        "required_inputs": ["pattern"],
        "optional_inputs": {"path": ".", "file_glob": "*"},
        "risk_level": "low",
        "fallbacks": ["directory_tree"],
        "_fn": _file_search,
    },
    "git_status": {
        "name": "git_status",
        "goal": "Mostra branch, file modificati e log recente",
        "description": "S763: git status --short + log --oneline -5. Read-only.",
        "required_inputs": [],
        "optional_inputs": {"cwd": "."},
        "risk_level": "low",
        "fallbacks": [],
        "_fn": _git_status,
    },
    "git_clone": {
        "name": "git_clone",
        "goal": "Clona un repository GitHub/GitLab",
        "description": "S763: git clone <url> [directory]. depth per shallow clone. Timeout 120s.",
        "required_inputs": ["url"],
        "optional_inputs": {"directory": "", "depth": 0},
        "risk_level": "medium",
        "fallbacks": [],
        "_fn": _git_clone,
    },
    "git_diff": {
        "name": "git_diff",
        "goal": "Mostra differenze tra modifiche correnti e ultimo commit",
        "description": "S763: git diff [--cached]. Stat + testo diff max 3000 chars. Read-only.",
        "required_inputs": [],
        "optional_inputs": {"cwd": ".", "staged": False},
        "risk_level": "low",
        "fallbacks": [],
        "_fn": _git_diff,
    },
    "get_image": {
        "name": "get_image",
        "goal": "Genera un immagine con AI (alias di generate_image)",
        "description": "S-GAP14: alias di generate_image via Pollinations.ai.",
        "required_inputs": ["prompt"],
        "optional_inputs": {"width": 512, "height": 512},
        "risk_level": "low",
        "fallbacks": ["generate_image"],
        "_fn": _generate_image,
    },
    "create_project": {
        "name": "create_project",
        "goal": "Crea struttura scaffolding progetto (alias di scaffold_project)",
        "description": "S-GAP14: alias di scaffold_project. Genera struttura cartelle + file base.",
        "required_inputs": ["framework", "project_name"],
        "optional_inputs": {"target_dir": "/tmp"},
        "risk_level": "low",
        "fallbacks": ["scaffold_project"],
        "_fn": _scaffold_project,
    },
    "create_chart": {
        "name": "create_chart",
        "goal": "Genera un grafico (bar, line, pie, scatter) da dati numerici",
        "description": (
            "S-GAP15: usa matplotlib (PNG base64) se disponibile, fallback SVG testuale. "
            "chart_type: bar|line|pie|scatter. data: dict {label: value} o labels+values list. "
            "title/x_label/y_label opzionali. Restituisce {type, format, data (base64), note}."
        ),
        "required_inputs": [],
        "optional_inputs": {
            "chart_type": "bar",
            "data": None,
            "title": "",
            "x_label": "",
            "y_label": "",
            "labels": None,
            "values": None,
        },
        "risk_level": "low",
        "fallbacks": [],
        "_fn": _create_chart,
    },
    "recall": {
        "name": "recall",
        "goal": "Cerca informazioni salvate in memoria dall\'agente",
        "description": "S-GAP13: cerca in agentMemory per query. Restituisce entries corrispondenti.",
        "required_inputs": ["query"],
        "optional_inputs": {"limit": 5},
        "risk_level": "low",
        "fallbacks": [],
        "_fn": _recall,
    },
    "list_files": {
        "name": "list_files",
        "goal": "Elenca file e cartelle in una directory",
        "description": "S-GAP13: os.listdir/os.walk. recursive=True per albero completo. Max 100 items.",
        "required_inputs": [],
        "optional_inputs": {"path": ".", "recursive": False, "max_items": 100},
        "risk_level": "low",
        "fallbacks": ["directory_tree"],
        "_fn": _list_files,
    },
    "diff_text": {
        "name": "diff_text",
        "goal": "Confronta due testi e mostra le differenze (unified diff)",
        "description": "S-GAP13: difflib.unified_diff. Restituisce patch testo + conteggio righe aggiunte/rimosse.",
        "required_inputs": ["text_a", "text_b"],
        "optional_inputs": {"context_lines": 3},
        "risk_level": "low",
        "fallbacks": [],
        "_fn": _diff_text,
    },
    "validate_json": {
        "name": "validate_json",
        "goal": "Valida se una stringa e JSON valido (con schema opzionale)",
        "description": "S-GAP13: json.loads + jsonschema opzionale. Restituisce valid, type, errori.",
        "required_inputs": ["json_str"],
        "optional_inputs": {"schema": None},
        "risk_level": "low",
        "fallbacks": [],
        "_fn": _validate_json,
    },
    "lint_code": {
        "name": "lint_code",
        "goal": "Analisi statica del codice (Python/JS/TS/JSON)",
        "description": "S-GAP13: wrapper di api.linter.lint_code. Auto-detect linguaggio da estensione file.",
        "required_inputs": ["content"],
        "optional_inputs": {"language": "auto", "path": ""},
        "risk_level": "low",
        "fallbacks": [],
        "_fn": _lint_code_tool,
    },
    "git_sync_vfs": {
        "name": "git_sync_vfs",
        "goal": "Sincronizza file VFS sessione su branch GitHub dedicato (snapshot persistenza)",
        "description": (
            "RF-1: Commit atomico VFS→GitHub via Git Data API (blob→tree→commit→PATCH ref). "
            "files: dict path→content dei file da sincronizzare (obbligatorio). "
            "branch: branch target (default: agent-state — creato auto se mancante). "
            "message: messaggio commit. repo: owner/repo (default: env GITHUB_REPO). "
            "Usa GITHUB_TOKEN da env. Zero clone locale."
        ),
        "required_inputs": ["files"],
        "optional_inputs": {"branch": "agent-state", "message": "chore(vfs): auto-sync session", "repo": ""},
        "risk_level": "medium",
        "fallbacks": ["git_push"],
        "_fn": _git_sync_vfs,
    },
    "git_push": {
        "name": "git_push",
        "goal": "Invia i commit al repository remoto (git push)",
        "description": (
            "S-GAP12: git push <remote> [<branch>]. "
            "remote: nome del remote (default: origin). "
            "branch: branch da pushare (default: branch corrente). "
            "Timeout 65s. Risk high: modifica il repository remoto."
        ),
        "required_inputs": [],
        "optional_inputs": {"remote": "origin", "branch": "", "cwd": "."},
        "risk_level": "high",
        "fallbacks": ["git_commit"],
        "_fn": _git_push,
    },
    "git_commit": {
        "name": "git_commit",
        "goal": "Esegue git add + commit delle modifiche correnti (e push opzionale)",
        "description": "S763: git add -A + commit -m <message>. push=True per git push. Risk medium.",
        "required_inputs": ["message"],
        "optional_inputs": {"cwd": ".", "push": False, "add_all": True},
        "risk_level": "medium",
        "fallbacks": [],
        "_fn": _git_commit,
    },
    "npm_install": {
        "name": "npm_install",
        "goal": "Installa dipendenze Node.js (npm/pnpm/yarn)",
        "description": "S763: auto-detecta manager da lockfile. Timeout 120s. Risk medium.",
        "required_inputs": [],
        "optional_inputs": {"cwd": ".", "manager": "auto", "args": ""},
        "risk_level": "medium",
        "fallbacks": [],
        "_fn": _npm_install,
    },
    "npm_run": {
        "name": "npm_run",
        "goal": "Esegue uno script npm (build, test, lint, typecheck, dev)",
        "description": "S763: npm/pnpm/yarn run <script>. Auto-detecta manager. Timeout 120s.",
        "required_inputs": ["script"],
        "optional_inputs": {"cwd": ".", "manager": "auto"},
        "risk_level": "medium",
        "fallbacks": [],
        "_fn": _npm_run,
    },
    "pip_install": {
        "name": "pip_install",
        "goal": "Installa pacchetti Python con pip",
        "description": "S763: pip install <packages>. Stringa spazio/virgola separata. Timeout 120s.",
        "required_inputs": ["packages"],
        "optional_inputs": {"upgrade": False, "user": False},
        "risk_level": "medium",
        "fallbacks": [],
        "_fn": _pip_install,
    },
    "type_check": {
        "name": "type_check",
        "goal": "Verifica tipi TypeScript (tsc --noEmit) o Python (mypy)",
        "description": "S763: auto-detecta checker da tsconfig.json/pyproject.toml. Timeout 60s.",
        "required_inputs": [],
        "optional_inputs": {"path": ".", "checker": "auto", "strict": False},
        "risk_level": "low",
        "fallbacks": [],
        "_fn": _type_check,
    },
    

    "trigger_webhook": {
        "name": "trigger_webhook",
        "goal": "Invia notifica HTTP a un webhook esterno (Zapier, Make, n8n, Discord, Slack, CI/CD...)",
        "description": (
            "P17-F4: HTTP POST/GET/PUT con payload JSON a URL esterno. "
            "Usa per automazioni, notifiche, CI/CD trigger, testing endpoint. "
            "Parametri: url (obbligatorio), payload (dict JSON), method (POST/GET/PUT), "
            "headers (dict), timeout (max 10s). "
            "Allowlist host via WEBHOOK_ALLOWED_HOSTS env (vuoto = tutti consentiti in dev)."
        ),
        "required_inputs": ["url"],
        "optional_inputs": {"payload": None, "method": "POST", "headers": None, "timeout": 10},
        "risk_level": "medium",
        "fallbacks": ["read_page"],
        "_fn": _trigger_webhook,
    },
    "write_and_check": {
        "name": "write_and_check",
        "goal": "Scrive un file e verifica con lint (Python/JS/TS/JSON/CSS) in un unico step",
        "description": (
            "P24-F1 macro: write_file + lint_code automatico se linguaggio supportato. "
            "path + content obbligatori. language: auto|py|js|ts|json|css. "
            "run_lint: True. Restituisce written, lint report, has_errors."
        ),
        "required_inputs": ["path", "content"],
        "optional_inputs": {"language": "auto", "run_lint": True},
        "risk_level": "medium",
        "fallbacks": ["write_file"],
        "_fn": _write_and_check,
    },
    "bulk_write_files": {
        "name": "bulk_write_files",
        "goal": "Scrive più file in parallelo in un solo step (dict path→content)",
        "description": (
            "P24-F1 macro: write_file x N con asyncio.gather. "
            "files: dict {path: content} obbligatorio. "
            "stop_on_error: False (continua su errori). run_lint: False. "
            "Restituisce written[], failed[], written_count."
        ),
        "required_inputs": ["files"],
        "optional_inputs": {"stop_on_error": False, "run_lint": False},
        "risk_level": "medium",
        "fallbacks": ["write_file"],
        "_fn": _bulk_write_files,
    },
    "fetch_and_extract": {
        "name": "fetch_and_extract",
        "goal": "Fetcha URL e restituisce struttura compatta: titolo, testo, parole, link",
        "description": (
            "P24-F1 macro: read_page + estrazione strutturata. "
            "url obbligatorio. extract: all|text|links|summary. "
            "Ritorna title, word_count, paragraphs[], excerpt, links[] (se extract=all/links)."
        ),
        "required_inputs": ["url"],
        "optional_inputs": {"extract": "all"},
        "risk_level": "low",
        "fallbacks": ["read_page"],
        "_fn": _fetch_and_extract,
    },
    "read_multiple_files": {
        "name": "read_multiple_files",
        "goal": "Legge più file in parallelo e restituisce i loro contenuti aggregati",
        "description": (
            "P24-F1 macro: read_file x N con asyncio.gather. "
            "paths: lista percorsi. max_chars_per_file: 4000 (default). "
            "Ritorna files {path: content}, errors {path: error}, read_count."
        ),
        "required_inputs": ["paths"],
        "optional_inputs": {"max_chars_per_file": 4000},
        "risk_level": "low",
        "fallbacks": ["read_file"],
        "_fn": _read_multiple_files,
    },
    "notion_rw": {
        "name": "notion_rw",
        "goal": "Leggi, scrivi e cerca pagine Notion (search/read/write/append)",
        "description": (
            "P24-F2: Integrazione Notion. "
            "action=search: cerca pagine per testo (query). "
            "action=read: contenuto pagina (page_id). "
            "action=write: crea pagina (parent_id+title+content). "
            "action=append: aggiunge a pagina esistente (page_id+content). "
            "Richiede NOTION_TOKEN env var nel backend."
        ),
        "required_inputs": ["action"],
        "optional_inputs": {"query": "", "page_id": "", "parent_id": "", "title": "", "content": "", "max_results": 5},
        "risk_level": "medium",
        "fallbacks": [],
        "_fn": _notion_rw,
    },
    "jina_fetch": {
        "name": "jina_fetch",
        "goal": "Leggi qualsiasi URL (incluse SPA/React/Next.js) tramite Jina Reader",
        "description": (
            "P24-F3: Jina Reader fallback per SPA e siti JavaScript. "
            "Restituisce testo markdown pulito da qualsiasi URL, "
            "inclusi siti renderizzati lato client (React/Vue/Angular/Next.js). "
            "url obbligatorio. query: filtra per rilevanza. "
            "target_selector: CSS selector da isolare. "
            "remove_selector: CSS da rimuovere (ads/nav). "
            "Funziona senza API key; con JINA_API_KEY limiti più alti."
        ),
        "required_inputs": ["url"],
        "optional_inputs": {"query": "", "target_selector": "", "remove_selector": "", "max_length": 12000},
        "risk_level": "low",
        "fallbacks": [],
        "_fn": _jina_fetch,
    },
    "python_analyze": {
        "name": "python_analyze",
        "goal": "Analizza codice Python: sintassi, struttura, metriche e suggerimenti",
        "description": (
            "P30-B1: Analisi statica Python in-process — zero exec_engine, <5ms. "
            "code: stringa Python (obbligatorio). filename: nome file per errori (default '<string>'). "
            "Ritorna: syntax_ok (bool), errors[] (type/message/line/col), "
            "complexity{total_lines/functions/classes/imports/max_nesting}, "
            "suggestions[] (max 5 actionable), summary (stringa leggibile). "
            "Usare per: verificare sintassi prima di exec, analizzare qualità codice, "
            "ottenere metriche strutturali, suggerire refactoring."
        ),
        "required_inputs": ["code"],
        "optional_inputs": {"filename": "<string>", "content": ""},  # GAP-1: alias per 'code' (frontend compat)
        "risk_level": "low",
        "fallbacks": [],
        "_fn": _python_analyze,
    },
}