agents/unified_loop_tools.py

"""unified_loop_tools.py — DirectToolsMixin: tool execution layer.
Estratto da unified_loop.py per ridurre il file principale da 2541 a ~2000 righe.
Contiene (nell'ordine originale del file):
  - Regex class attrs: meteo, URL, ricerca, immagini, calcolo
  - Helper: _extract_city / _extract_search_query / _extract_calc_expr
  - _run_direct_tools: layer deterministico parallelo via TOOL_REGISTRY (S193/S419)
  - _FALSE_CLAIM_RE / _REALTIME_GOAL_RE / _validate_claims: anti-hallucination (S428)
  - _TOOL_NEEDED_RE / _needs_tools / _SIMPLE_CONV_RE / _is_simple_query: routing (S402)
Invariante B1: nessun corpo duplicato con unified_loop.py.
Python MRO garantisce che self.xxx funzioni per attr definite su UnifiedAgentLoop.
"""
from __future__ import annotations
import asyncio
import os
import re
from typing import Any
import logging
try:
    from api.state import record_timing as _rtc_global  # telemetria tool call
except ImportError:
    _rtc_global = None  # state module non ancora disponibile al boot

_logger = logging.getLogger("agents.unified_loop_tools")
# StepCallback centralizzato in unified_loop_types.py (P20-TD1 Fase 1)
# S-FIX-IMPORT: aggiunto _maybe_await mancante che causava crash nel tool layer
from agents.unified_loop_types import StepCallback, _maybe_await
class DirectToolsMixin:
    # ── Direct tool execution (S193) ─────────────────────────────────────────
    # Chiama TOOL_REGISTRY direttamente, senza smolagents, senza LLM per routing.
    # Deterministico, veloce, testabile. Restituisce i risultati come stringa
    # pronta per essere iniettata nel prompt LLM.
    _WEATHER_INTENT_RE = re.compile(
        # S390-B-O: aggiunto 'temperature' (inglese) + 'forecast' come sinonimi weather
        # S427: aggiunti fenomeni meteo, allerte, condizioni IT/EN
        r"\b(meteo|temperatura|temperature|temp\s*a\b|clima|weather|previsioni|forecast|"
        r"che\s+tempo\s+fa|quanto\s+fa\s+freddo|quanto\s+fa\s+caldo|gradi\s+a\b|"
        r"piove|sta\s+piovendo|nevica|neve|pioggia|temporale|grandine|"
        r"nebbia|umidità|vento|allerta\s+meteo|allerta\s+rossa|allerta\s+arancione|"
        r"ondata\s+di\s+caldo|ondata\s+di\s+freddo|gelate|gelo|"
        r"rain|raining|snow|snowing|fog|humid|wind|windy|storm|thunderstorm|hail|"
        r"sunny|cloudy|overcast|uv\s+index|heat\s+wave|cold\s+snap|frost)\b",
        re.IGNORECASE,
    )
    # S385: improved city extraction — catches bare patterns like "che tempo fa a Roma?"
    # S390-B-O: aggiunti trigger 'temperature\s+in' e 'weather\s+(?:forecast\s+)?(?:in|at|for)'
    _CITY_RE = re.compile(
        r"(?:meteo|temperatura|temperature|temp(?:eratura)?\s+(?:a|in)|"
        r"(?:che\s+)?tempo\s+(?:\w+\s+){0,2}(?:fa\s+)?(?:a|in)|"
        r"com['\u2019]è\s+il\s+tempo\s+a|com['\u2019]è\s+il\s+meteo\s+a|"
        r"clima\s+(?:a|in)|weather\s+(?:forecast\s+)?(?:in|at|for)|"
        r"temperature\s+(?:in|at|for)|forecast\s+(?:for|in)|"
        r"previsioni\s+(?:per|a|in)|gradi\s+(?:a|in))"
        r"\s+(?:a\s+|in\s+|per\s+)?([A-Za-z\xc0-\xff][A-Za-z\xc0-\xff\s]{1,25}?)"
        # S390-B-I: aggiunti terminatori inglesi (today/now/tomorrow/currently/right now)
        r"(?:\?|$|\s*[,\.]|\s+adesso|\s+ora|\s+oggi|\s+domani|\s+attuale|\s+corrente"
        r"|\s+today|\s+now|\s+tomorrow|\s+currently|\s+right\s+now)",
        re.IGNORECASE,
    )
    _URL_RE = re.compile(r"https?://[^\s\)\}\]>]+", re.IGNORECASE)
    _SEARCH_INTENT_RE = re.compile(
        r"\b(cerca|search|trova|find|googla|google|duckduckgo|bing|research|investiga|indaga|"
        r"fammi\s+sapere|dimmi\s+di\s+più\s+su|informazioni\s+su|info\s+su|news\s+su|notizie\s+su|"
        r"chi\s+è|cos['\u2019]è|dove\s+si\s+trova|quando\s+è\s+successo|perché\s+il|storia\s+di|"
        r"tell\s+me\s+about|who\s+is|what\s+is|where\s+is|when\s+did|why\s+is|history\s+of|"
        r"latest\s+on|ultime\s+su|prezzo\s+di|valore\s+di|quotazione\s+di|stock\s+price\s+of|"
        r"crypto|bitcoin|ethereum|market\s+cap|capitalizzazione)\b",
        re.IGNORECASE,
    )
    _IMAGE_INTENT_RE = re.compile(
        r"\b(genera|crea|disegna|illustra|fai|mostra|fammi\s+un[a']?|visualizza|produce|render|paint|sketch|"
        r"immagine|foto|illustrazione|ritratto|paesaggio|logo|icona|disegno|grafica|"
        r"image|photo|illustration|portrait|landscape|drawing|graphic|art|artwork)\b",
        re.IGNORECASE,
    )
    _CALC_INTENT_RE = re.compile(
        r"\b(calcola|quanto\s+fa|risultato\s+di|compute|calculate|math|matematica|operazione|"
        r"somma|sottrai|moltiplica|dividi|percentuale|radice|potenza|"
        r"sum|add|subtract|multiply|divide|percentage|root|power)\b",
        re.IGNORECASE,
    )
    def _extract_city(self, goal: str) -> str:
        m = self._CITY_RE.search(goal)
        if m:
            candidate = m.group(1).strip()
            if len(candidate) > 1 and candidate not in {"in", "nel", "un", "il"}:
                return candidate
        return "."
    def _extract_search_query(self, goal: str) -> str:
        q = re.sub(self._SEARCH_INTENT_RE, "", goal, flags=re.IGNORECASE).strip()
        return q or goal
    def _extract_calc_expr(self, goal: str) -> str:
        m = re.search(r'[\d\s\+\-\*\/\^\(\)\.]+', goal)
        return m.group(0).strip() if m else ""

    def _extract_dir_path(self, goal: str) -> str:
        """Extract a safe relative directory path, defaulting to the tool root."""
        m = re.search(
            r"(?:di|in|dentro|in\s+path|nel\s+path|directory|folder|cartella)\s+"
            r"['\"]?([./\w\-]+/[./\w\-]*|[./\w\-]+)['\"]?",
            goal, re.IGNORECASE,
        )
        if m:
            candidate = m.group(1).strip().rstrip("/")
            if candidate not in {"di", "in", "nel", "nella"}:
                return candidate
        return "."

    def _extract_file_pattern(self, goal: str) -> str:
        """Extract the search pattern without changing the registry's FS jail."""
        m = re.search(
            r"(?:grep\s+|cerca\s+(?:la\s+stringa\s+)?|trova\s+(?:la\s+stringa\s+)?|"
            r"search\s+for\s+|find\s+in\s+files\s+)['\"]?([^\s'\"?,]{2,80})['\"]?",
            goal, re.IGNORECASE,
        )
        return m.group(1).strip() if m else ""

    def _extract_git_cwd(self, goal: str) -> str:
        """Extract the requested git working directory, defaulting to root."""
        m = re.search(
            r"(?:in|nel\s+repo|nel\s+repository|in\s+path)\s+['\"]?([./\w\-]+)['\"]?",
            goal, re.IGNORECASE,
        )
        if m:
            candidate = m.group(1).strip()
            if len(candidate) > 1 and candidate not in {"in", "nel", "un", "il"}:
                return candidate
        return "."
    async def _run_direct_tools(self, goal: str, on_step: StepCallback | None = None) -> tuple[str, int, int, int]:
        """
        S193: Esegue tool direttamente via TOOL_REGISTRY senza smolagents o LLM per routing.
        Returns: 4-tuple (results_str, n_called, n_success, n_errors).
            results_str: stringa reale da iniettare nel prompt (join di tutti i tool output)
            n_called:    numero totale di tool chiamati
            n_success:   numero di tool completati con successo
            n_errors:    numero di tool falliti
        """
        # FIX-TOOL-01: usare i package reali del backend; i moduli indicati dal
        # precedente restore non esistono e interrompevano il direct-tools layer.
        from tools.registry import TOOL_REGISTRY
        from api.speculative import get_speculative_result as _speculative_result

        results: list[str] = []
        n_called = 0
        n_success = 0
        n_errors = 0
        TOOL_TIMEOUT = 25

        # Governor per singolo run: conserva budget adattivo e deduplicazione.
        _gov_called: set[str] = set()
        _gov_total = 0
        _tok_budget_gov = self._max_tokens_for_goal(goal)
        _gov_max_calls = 9 if _tok_budget_gov >= 6144 else 7 if _tok_budget_gov >= 4096 else 6

        def _gov_check(tool_name: str, key_arg: str) -> bool:
            nonlocal _gov_total
            if _gov_total >= _gov_max_calls:
                return False
            signature = f"{tool_name}:{key_arg[:150]}"
            if signature in _gov_called:
                return False
            _gov_called.add(signature)
            _gov_total += 1
            return True

        def _spec_hit(tool_name: str, args: dict[str, Any]) -> str | None:
            try:
                return _speculative_result(goal, tool_name, args)
            except Exception:
                # Cache speculativa opzionale: mai bloccare l'esecuzione reale.
                return None

        # S419: esegui i tool eligible in parallelo con asyncio.gather
        # Pre-check intent (sincrono) → costruisce lista coroutine → gather
        url_m = self._URL_RE.search(goal)
        async def _t_get_weather() -> str | None:
            if not self._WEATHER_INTENT_RE.search(goal):
                return None
            city = self._extract_city(goal)
            if not _gov_check("get_weather", city):
                return None
            try:
                if on_step:
                    await _maybe_await(on_step({"action": "tool_start", "status": "running",
                        "title": "Meteo", "explanation": f"Recupero meteo per {city}…"}))
                _sc = _spec_hit("get_weather", {"city": city})
                if _sc is not None:
                    return _sc
                _t0 = asyncio.get_event_loop().time()
                r = await asyncio.wait_for(TOOL_REGISTRY["get_weather"]["_fn"](city=city), timeout=TOOL_TIMEOUT)
                try:
                    from api.state import record_timing as _rtc; _rtc("tool_call", (asyncio.get_event_loop().time() - _t0) * 1000)
                except Exception as _e: _logger.debug('[timing/record_timing] %s', _e)
                if "temp_c" in r:
                    _wdesc = {
                        0: "cielo sereno", 1: "prevalentemente sereno", 2: "parzialmente nuvoloso", 3: "coperto",
                        45: "nebbia", 48: "nebbia con brina", 51: "pioviggine leggera", 53: "pioviggine moderata",
                        55: "pioviggine intensa", 61: "pioggia leggera", 63: "pioggia moderata", 65: "pioggia forte",
                        71: "nevicata leggera", 73: "nevicata moderata", 75: "nevicata forte", 80: "rovesci leggeri",
                        81: "rovesci moderati", 82: "rovesci violenti", 95: "temporale", 96: "temporale con grandine",
                    }
                    wcode = r.get("code"); temp_c = r.get("temp_c"); wind_kmh = r.get("wind_kmh")
                    try:
                        desc = _wdesc.get(int(wcode), f"codice {wcode}") if wcode is not None else "N/D"
                    except (TypeError, ValueError):
                        desc = "N/D"
                    return (
                        f"[METEO REALE — {r['city']}, {r.get('country', '')}]\n"
                        f"Temperatura attuale: {f'{temp_c}°C' if temp_c is not None else 'N/D'}\n"
                        f"Vento: {f'{wind_kmh} km/h' if wind_kmh is not None else 'N/D'}\n"
                        f"Condizioni: {desc}"
                    )
                return f"[get_weather: errore — {r['error'][:300]}]"
            except asyncio.TimeoutError:
                return f"[get_weather: timeout {TOOL_TIMEOUT}s]"
            except Exception as exc:
                return f"[get_weather: errore — {str(exc)[:300]}]"
        async def _t_read_page() -> str | None:
            if not url_m:
                return None
            url = url_m.group(0)
            if not _gov_check("read_page", url):
                return None
            try:
                _sc = _spec_hit("read_page", {"url": url})
                if _sc is not None:
                    return _sc
                if on_step:
                    await _maybe_await(on_step({"action": "tool_start", "status": "running",
                        "title": "Lettura pagina", "explanation": f"Leggo {url[:60]}…"}))
                _t0 = asyncio.get_event_loop().time()
                r = await asyncio.wait_for(TOOL_REGISTRY["read_page"]["_fn"](url=url), timeout=TOOL_TIMEOUT)
                try:
                    from api.state import record_timing as _rtc; _rtc("tool_call", (asyncio.get_event_loop().time() - _t0) * 1000)
                except Exception as _e: _logger.debug('[timing/record_timing] %s', _e)
                if r.get("content"):
                    return (f"[PAGINA REALE: {url}]\n(status {r.get('status', '?')})\n{r['content'][:3000]}")
                return f"[read_page: errore — {r.get('error', 'nessun contenuto')[:300]}]"
            except asyncio.TimeoutError:
                return f"[read_page: timeout {TOOL_TIMEOUT}s]"
            except Exception as exc:
                return f"[read_page: errore — {str(exc)[:300]}]"
        async def _t_calculate() -> str | None:
            if url_m or not self._CALC_INTENT_RE.search(goal):
                return None
            expr = self._extract_calc_expr(goal)
            if not expr or not _gov_check("calculate", expr):
                return None
            try:
                if on_step:
                    await _maybe_await(on_step({"action": "tool_start", "status": "running",
                        "title": "Calcolo", "explanation": f"Calcolo: {expr[:60]}"}))
                _sc = _spec_hit("calculate", {"expression": expr})
                if _sc is not None:
                    return _sc
                _t0 = asyncio.get_event_loop().time()
                r = await asyncio.wait_for(TOOL_REGISTRY["calculate"]["_fn"](expression=expr), timeout=8)
                try:
                    from api.state import record_timing as _rtc; _rtc("tool_call", (asyncio.get_event_loop().time() - _t0) * 1000)
                except Exception as _e: _logger.debug('[timing/record_timing] %s', _e)
                if "result" in r:
                    return f"[CALCOLO REALE]\n{r['expression']} = {r['result']}"
                return f"[calculate: errore — {r.get('error', '?')[:300]}]"
            except asyncio.TimeoutError:
                return "[calculate: timeout]"
            except Exception as exc:
                return f"[calculate: errore — {str(exc)[:300]}]"
        async def _t_web_search() -> str | None:
            if not self._SEARCH_INTENT_RE.search(goal):
                return None
            query = self._extract_search_query(goal)
            if not query or not _gov_check("web_search", query):
                return None
            try:
                if on_step:
                    await _maybe_await(on_step({"action": "tool_start", "status": "running",
                        "title": "Ricerca web", "explanation": f"Cerco: {query[:60]}…"}))
                _sc = _spec_hit("web_search", {"query": query})
                if _sc is not None:
                    return _sc
                _t0 = asyncio.get_event_loop().time()
                r = await asyncio.wait_for(TOOL_REGISTRY["web_search"]["_fn"](query=query, max_results=5), timeout=TOOL_TIMEOUT)
                try:
                    from api.state import record_timing as _rtc; _rtc("tool_call", (asyncio.get_event_loop().time() - _t0) * 1000)
                except Exception as _e: _logger.debug('[timing/record_timing] %s', _e)
                hits = r.get("results", [])
                if hits:
                    _out = [f"[RICERCA WEB REALE: {query}]"]
                    for h in hits:
                        _out.append(f"• {h['title']} ({h['url']}): {h['snippet']}")
                    return "\n".join(_out)
                return f"[web_search: nessun risultato per '{query}']"
            except asyncio.TimeoutError:
                return f"[web_search: timeout {TOOL_TIMEOUT}s]"
            except Exception as exc:
                return f"[web_search: errore — {str(exc)[:300]}]"
        async def _t_generate_image() -> str | None:
            if not self._IMAGE_INTENT_RE.search(goal):
                return None
            _img_prompt = re.sub(
                r"^.*?(?:genera|crea|disegna|illustra|fai|mostra).*?(?:immagine|foto|illustrazione|di|un[a']?|del?la?|del?l[o']?)\s*",
                "", goal, flags=re.IGNORECASE
            ).strip() or goal
            if not _gov_check("generate_image", _img_prompt):
                return None
            try:
                if on_step:
                    await _maybe_await(on_step({"action": "tool_start", "status": "running",
                        "title": "Generazione immagine", "explanation": f"Genero: {_img_prompt[:60]}…"}))
                _sc = _spec_hit("generate_image", {"prompt": _img_prompt[:600]})
                if _sc is not None:
                    return _sc
                _t0 = asyncio.get_event_loop().time()
                r = await asyncio.wait_for(TOOL_REGISTRY["generate_image"]["_fn"](prompt=_img_prompt[:600]), timeout=12)
                try:
                    from api.state import record_timing as _rtc; _rtc("tool_call", (asyncio.get_event_loop().time() - _t0) * 1000)
                except Exception as _e: _logger.debug('[timing/record_timing] %s', _e)
                img_url = r.get("url", "")
                if img_url:
                    return (
                        f"[IMMAGINE AI GENERATA]\n"
                        f"URL: {img_url}\n"
                        f"Prompt usato: {r.get('prompt', _img_prompt)[:200]}\n"
                        f"Dimensioni: {r.get('width')}x{r.get('height')} px"
                    )
                return "[generate_image: nessun URL restituito]"
            except asyncio.TimeoutError:
                return "[generate_image: timeout — provider non raggiungibile]"
            except Exception as exc:
                return f"[generate_image: errore — {str(exc)[:300]}]"
        async def _t_run_python() -> str | None:
            _RUN_CODE_RE = re.compile(
                r"\b(?:run\s+(?:python\s+)?code|esegui\s+(?:\w+\s+){0,2}codice|"
                r"execute\s+(?:python\s+)?code|lancia\s+(?:il\s+)?codice|"
                r"esegui\s+(?:questo\s+|il\s+)?(?:script|programma))\b",
                re.IGNORECASE,
            )
            if not _RUN_CODE_RE.search(goal):
                return None
            _code_m = re.search(r"[:：]\s*(.+)$", goal, re.DOTALL)
            _code = _code_m.group(1).strip() if _code_m else goal
            _code = re.sub(r"^```(?:python)?\s*|\s*```$", "", _code.strip(), flags=re.DOTALL).strip()
            if not _code or len(_code) <= 3 or not _gov_check("run_python", _code[:80]):
                return None
            try:
                if on_step:
                    await _maybe_await(on_step({"action": "tool_start", "status": "running",
                        "title": "Esecuzione codice Python", "explanation": "Eseguo il codice in sandbox…"}))
                _sc = _spec_hit("run_python", {"code": _code[:400]})
                if _sc is not None:
                    return _sc
                _t0 = asyncio.get_event_loop().time()
                r = await asyncio.wait_for(TOOL_REGISTRY["run_python"]["_fn"](code=_code), timeout=18)
                try:
                    from api.state import record_timing as _rtc; _rtc("tool_call", (asyncio.get_event_loop().time() - _t0) * 1000)
                except Exception as _e: _logger.debug('[timing/record_timing] %s', _e)
                if r.get("returncode", -1) == 0 and r.get("stdout"):
                    _out = (
                        "[CODICE PYTHON ESEGUITO]\n"
                        f"```python\n{_code[:500]}\n```\n"
                        f"Output:\n```\n{r['stdout'][:1500]}\n```"
                    )
                    return _out
                return f"[run_python: errore — {r.get('stderr', 'ignoto')[:300]}]"
            except asyncio.TimeoutError:
                return "[run_python: timeout 18s]"
            except Exception as exc:
                return f"[run_python: errore — {str(exc)[:300]}]"
        async def _t_web_research() -> str | None:
            _RESEARCH_RE = re.compile(r"\b(ricerca\s+approfondita|deep\s+research|investigazione|analisi\s+dettagliata)\b", re.IGNORECASE)
            if not _RESEARCH_RE.search(goal):
                return None
            query = self._extract_search_query(goal)
            if not _gov_check("web_research", query):
                return None
            try:
                if on_step:
                    await _maybe_await(on_step({"action": "tool_start", "status": "running",
                        "title": "Ricerca approfondita", "explanation": f"Analisi dettagliata su: {query[:60]}…"}))
                _t0 = asyncio.get_event_loop().time()
                r = await asyncio.wait_for(TOOL_REGISTRY["web_research"]["_fn"](query=query), timeout=45)
                try:
                    from api.state import record_timing as _rtc; _rtc("tool_call", (asyncio.get_event_loop().time() - _t0) * 1000)
                except Exception as _e: _logger.debug('[timing/record_timing] %s', _e)
                if r.get("report"):
                    return f"[RICERCA APPROFONDITA REALE: {query}]\n\n{r['report'][:4000]}"
                return f"[web_research: errore — {r.get('error', 'nessun report')[:300]}]"
            except asyncio.TimeoutError:
                return "[web_research: timeout 45s]"
            except Exception as exc:
                return f"[web_research: errore — {str(exc)[:300]}]"
        async def _t_directory_tree() -> str | None:
            _TREE_RE = re.compile(r"\b(albero|struttura|directory\s+tree|files?|cartell[ae])\b", re.IGNORECASE)
            if not _TREE_RE.search(goal):
                return None
            _path = self._extract_dir_path(goal)
            if not _gov_check("directory_tree", _path):
                return None
            try:
                if on_step:
                    await _maybe_await(on_step({"action": "tool_start", "status": "running",
                        "title": "Struttura progetto", "explanation": f"Analisi directory: {_path}"}))
                r = await asyncio.wait_for(
                    TOOL_REGISTRY["directory_tree"]["_fn"](path=_path, max_depth=3), timeout=8
                )
                if r.get("ok") and r.get("tree"):
                    return f"[STRUTTURA PROGETTO REALE: '{_path}']\n{r['tree'][:2000]}"
                return f"[directory_tree: {r.get('error', 'nessun risultato')[:200]}]"
            except Exception as exc:
                return f"[directory_tree: errore — {str(exc)[:200]}]"
        async def _t_file_search() -> str | None:
            _SEARCH_RE = re.compile(r"\b(cerca\s+file|find\s+file|grep)\b", re.IGNORECASE)
            if not _SEARCH_RE.search(goal):
                return None
            _pattern = self._extract_file_pattern(goal)
            if not _pattern or not _gov_check("file_search", _pattern):
                return None
            _search_path = self._extract_dir_path(goal)
            try:
                if on_step:
                    await _maybe_await(on_step({"action": "tool_start", "status": "running",
                        "title": "Ricerca file", "explanation": f"Cerco '{_pattern}' nel codice…"}))
                r = await asyncio.wait_for(
                    TOOL_REGISTRY["file_search"]["_fn"](pattern=_pattern, path=_search_path), timeout=10
                )
                if r.get("ok"):
                    _matches = r.get("matches", [])
                    _out = [f"[FILE TROVATI: pattern='{_pattern}', {r.get('count', len(_matches))} occorrenze]"]
                    for match in _matches[:20]:
                        _out.append(f"{match.get('file', '?')}:{match.get('line', '?')}: {match.get('text', '')[:120]}")
                    return "\n".join(_out)
                return f"[file_search: {r.get('error', 'nessun risultato')[:200]}]"
            except Exception as exc:
                return f"[file_search: errore — {str(exc)[:200]}]"
        async def _t_get_news() -> str | None:
            _NEWS_RE = re.compile(r"\b(news|notizie|ultim[ae]\s+ora|breaking)\b", re.IGNORECASE)
            if not _NEWS_RE.search(goal):
                return None
            query = self._extract_search_query(goal)
            try:
                if on_step:
                    await _maybe_await(on_step({"action": "tool_start", "status": "running",
                        "title": "Notizie", "explanation": f"Cerco notizie su: {query[:60]}…"}))
                r = await asyncio.wait_for(TOOL_REGISTRY["get_news"]["_fn"](query=query), timeout=15)
                if r.get("news"):
                    _out = [f"[NOTIZIE REALI: {query}]"]
                    for n in r["news"][:5]:
                        _out.append(f"• {n['title']} ({n.get('source', '?')}): {n.get('description', '')[:150]}")
                    return "\n".join(_out)
                return "[get_news: nessuna notizia trovata]"
            except Exception as exc:
                return f"[get_news: errore — {str(exc)[:200]}]"
        async def _t_git_status() -> str | None:
            _GIT_RE = re.compile(r"\b(git|status|commit|branch|repo)\b", re.IGNORECASE)
            if not _GIT_RE.search(goal):
                return None
            _cwd = self._extract_git_cwd(goal)
            if not _gov_check("git_status", _cwd):
                return None
            try:
                if on_step:
                    await _maybe_await(on_step({"action": "tool_start", "status": "running",
                        "title": "Stato Git", "explanation": f"Controllo la repo in {_cwd}…"}))
                r = await asyncio.wait_for(
                    TOOL_REGISTRY["git_status"]["_fn"](cwd=_cwd), timeout=8
                )
                if r.get("ok"):
                    _out = [f"[STATO GIT REALE (branch: {r.get('branch', '?')})]"]
                    if r.get("status"):
                        _out.append(f"File modificati:\n{r['status'][:600]}")
                    if r.get("log"):
                        _out.append(f"Ultimi commit:\n{r['log'][:400]}")
                    return "\n".join(_out)
                return f"[git_status: {r.get('error', 'nessun risultato')[:200]}]"
            except Exception as exc:
                return f"[git_status: errore — {str(exc)[:200]}]"
        async def _t_analyze_python() -> str | None:
            # P30-B1: Analisi statica Python integrata nel tool layer
            if not self._ANALYZE_PY_RE.search(goal):
                return None
            _code = ""
            _m = self._PY_BLOCK_IN_GOAL_RE.search(goal)
            if _m: _code = _m.group(1).strip()
            if not _code: return None
            try:
                if on_step:
                    await _maybe_await(on_step({"action": "tool_start", "status": "running",
                        "title": "Analisi codice Python", "explanation": "Controllo sintassi e best practices…"}))
                from scripts.gap_map import analyze_python_code as _apc
                r = await asyncio.wait_for(_apc(_code), timeout=15)
                _out = ["[ANALISI PYTHON REALE]"]
                if r.get("errors"):
                    _out.append("❌ Errori rilevati:")
                    for _e in r["errors"]: _out.append(f"  - {_e}")
                else:
                    _out.append("✅ Nessun errore di sintassi rilevato.")
                if r.get("suggestions"):
                    _out.append("\n💡 Suggerimenti:")
                    for _s in r["suggestions"]: _out.append(f"  - {_s}")
                return "\n".join(_out)
            except asyncio.TimeoutError:
                return "[python_analyze: timeout]"
            except Exception as _exc:
                return f"[python_analyze: errore — {str(_exc)[:200]}]"
        # Esecuzione parallela
        _sem = asyncio.Semaphore(3)
        async def _sem_wrap(coro):
            if coro is None: return None
            async with _sem: return await coro
        _parallel_results = await asyncio.gather(
            _sem_wrap(_t_get_weather()),
            _sem_wrap(_t_read_page()),
            _sem_wrap(_t_calculate()),
            _sem_wrap(_t_web_search()),
            _sem_wrap(_t_generate_image()),
            _sem_wrap(_t_run_python()),
            _sem_wrap(_t_web_research()),
            _sem_wrap(_t_directory_tree()),
            _sem_wrap(_t_file_search()),
            _sem_wrap(_t_get_news()),
            _sem_wrap(_t_git_status()),
            _sem_wrap(_t_analyze_python()),
            return_exceptions=True,
        )
        for _pr in _parallel_results:
            if isinstance(_pr, str):
                results.append(_pr)
        # S428 Sprint1-Fix1: Tool Success Contract
        _REAL_DATA_PREFIXES = (
            "[RICERCA WEB REALE", "[METEO REALE", "[PAGINA REALE", "[CALCOLO REALE",
            "[IMMAGINE AI GENERATA", "[CODICE PYTHON ESEGUITO", "[RICERCA APPROFONDITA REALE",
            "[STRUTTURA PROGETTO REALE", "[RICERCA FILE REALE", "[NOTIZIE REALI",
            "[STATO GIT REALE", "[ANALISI PYTHON REALE"
        )
        for r_str in results:
            n_called += 1
            if any(r_str.startswith(p) for p in _REAL_DATA_PREFIXES):
                n_success += 1
            elif ": errore" in r_str or ": timeout" in r_str:
                n_errors += 1
        return ("\n\n".join(results), n_called, n_success, n_errors)
    # ── Claim Validation (S428 Sprint1-Fix3) ─────────────────────────────────
    # A failed live tool must never be represented as a successful live lookup.
    _FALSE_CLAIM_RE = re.compile(
        r"\b(ho\s+trovato(?:\s+che)?|ho\s+recuperato|ho\s+cercato\s+e\s+trovato|"
        r"dai\s+risultati(?:\s+della\s+ricerca)?|stando\s+ai\s+risultati|"
        r"i\s+risultati\s+(?:mostrano|indicano|confermano)|"
        r"la\s+ricerca\s+ha\s+(?:trovato|restituito)|"
        r"secondo\s+i\s+risultati|dalle\s+mie\s+ricerche|"
        r"I\s+found|the\s+results?\s+show|based\s+on\s+(?:the\s+)?results?|"
        r"according\s+to\s+(?:the\s+)?(?:search\s+)?results?)\b",
        re.IGNORECASE,
    )
    _REALTIME_GOAL_RE = re.compile(
        r"\b(notizie|news|ultime\s+notizie|cerca|ricerca\s+web|"
        r"weather|meteo|previsioni|temperatura|"
        r"bitcoin|ethereum|cambio\s+valuta|tasso|crypto|"
        r"versione\s+(?:attuale|corrente|recente)|aggiornamenti\s+su|release)\b",
        re.IGNORECASE,
    )

    @staticmethod
    def _validate_claims(
        response: str,
        n_success: int,
        n_errors: int,
        goal: str,
        false_claim_re: "re.Pattern[str]",
        realtime_goal_re: "re.Pattern[str]",
    ) -> str:
        """Add transparency when failed live tools are presented as successful."""
        if n_success > 0 or n_errors == 0:
            return response
        if not realtime_goal_re.search(goal):
            return response
        if not false_claim_re.search(response):
            return response
        disclaimer = (
            "\n\n---\n"
            "**Nota tecnica**: i servizi di ricerca in tempo reale non erano "
            "raggiungibili durante questa risposta. Le informazioni sopra provengono "
            "dal mio training e potrebbero non essere aggiornate. "
            "Per dati live consulta una fonte ufficiale."
        )
        return response + disclaimer
    _TOOL_NEEDED_RE = re.compile(
        r"\b(meteo|temperatura|weather|forecast|cerca|search|trova|find|googla|google|"
        r"immagine|foto|photo|image|disegna|draw|genera|create|calcola|calculate|math|"
        r"news|notizie|prezzo|quotazione|stock|crypto|bitcoin|albero|struttura|directory|"
        r"file|cartella|grep|python|esegui|run|execute|script|webhook|api|http|zapier|n8n)\b",
        re.IGNORECASE,
    )
    def _needs_tools(self, goal: str) -> bool:
        if len(goal) > 100: return True
        if bool(self._TOOL_NEEDED_RE.search(goal)): return True
        tech_keywords = ['file', 'directory', 'folder', 'script', 'api', 'json', 'data', 'analisi', 'fix', 'bug']
        if any(kw in goal.lower() for kw in tech_keywords): return True
        return False
    _SIMPLE_CONV_RE = re.compile(
        r"^(?:ciao|salve|hey\b|hi\b|hello\b|buongiorno|buonasera|buonanotte|"
        r"grazie(?:\s+mille)?|prego|perfetto|ottimo|esatto|capito|ok\b|bene\b|"
        r"come stai\??|come va\??|stai bene\??|chi sei\??|cosa sei\??|"
        r"sei (?:un[ao']?\s+)?(?:ai|bot|intelligenza artificiale|assistente)\??|"
        r"cosa (?:puoi fare|sai fare)\??|dimmi qualcosa di te|"
        r"bravo|benissimo|magnifico|fantastico|geniale|ottima risposta|"
        r"giusto|corretto|esattamente|d['\u2019]accordo|"
        r"capisco|ho capito|inteso|compreso|ricevuto|"
        r"s[iì] grazie|no grazie|va bene|va benissimo|"
        r"thanks|thank you|ty|thx|great|nice|perfect|exactly|understood|"
        r"got it|sure|right|agreed|makes sense|correct|good|"
        r"good morning|good evening|good night"
        r")\.?\s*[!?]?$",
        re.IGNORECASE,
    )
    _SIMPLE_MATH_RE = re.compile(
        r'^(?:(?:calcola|quanto\s+(?:fa|fanno|vale|valgono)|quant[oei]\s+(?:fa|fanno)|'
        r'dimmi\s+(?:solo\s+)?(?:il\s+)?(?:risultato|valore)\s+di|'
        r'compute|calculate|what(?:\'s|\s+is)\s+(?:the\s+(?:result\s+of\s+)?)?)\s*)?'
        r'[\d\s\+\-\*\/\^\(\)\.]+\s*[=?]?$',
        re.IGNORECASE,
    )
    _ANALYZE_PY_RE = re.compile(
        r"(?:analizza\s+(?:questo\s+)?(?:codice|script|programma)(?:\s+python)?"
        r"|analisi\s+(?:del\s+)?(?:codice|script)(?:\s+python)?"
        r"|check\s+(?:my\s+)?(?:python\s+)?(?:code|syntax|script)"
        r"|review\s+(?:my\s+)?(?:python\s+)?(?:code|script)"
        r"|syntax\s+check(?:\s+python)?"
        r"|verifica\s+(?:la\s+)?(?:sintassi|il\s+codice)(?:\s+python)?"
        r"|controlla\s+(?:il\s+)?(?:codice|sintassi)(?:\s+python)?"
        r"|esamina\s+(?:il\s+)?(?:codice|script)(?:\s+python)?)",
        re.IGNORECASE,
    )
    _PY_BLOCK_IN_GOAL_RE = re.compile(
        r"```(?:python|py)\s*\n([\s\S]+?)```",
        re.IGNORECASE,
    )
    _CODE_GOAL_RE = re.compile(r"\b(codice|script|programma|funzione|classe|modulo|libreria|package|repository|repo|git|github|branch|commit|pull\s+request|pr|merge|conflitto|conflict|test|unit\s+test|benchmark|profiling|debug|fix|bug|issue|refactor|ottimizzazione|optimization|typescript|javascript|python|rust|go|java|c\+\+|html|css|react|vue|angular|svelte|nextjs|vite|webpack|babel|eslint|prettier|npm|pnpm|yarn|docker|kubernetes|k8s|aws|gcp|azure|vercel|netlify|railway|supabase|firebase|database|sql|nosql|mongodb|postgresql|mysql|redis|api|rest|graphql|grpc|websocket|oauth|jwt|auth|sicurezza|security|crittografia|encryption|ai|llm|agente|agent|transformer|pytorch|tensorflow|scikit-learn|pandas|numpy|matplotlib|seaborn|plotly|fastapi|flask|django|express|koa|nest|spring|laravel|rails|symfony|phoenix|elixir|erlang|clojure|haskell|scala|kotlin|swift|objective-c|dart|flutter|react-native|expo|electron|tauri|capacitor|cordova|ionic|wasm|webassembly)\b", re.IGNORECASE)
    _CODE_RE = re.compile(r"```[\s\S]*?```")
    def _is_simple_query(self, goal: str) -> bool:
        g = goal.strip()
        if self._CODE_GOAL_RE.search(g) or self._CODE_RE.search(g):
            return False
        if len(g) <= 100 and self._SIMPLE_MATH_RE.match(g):
            return True
        if len(g) > 70 or self._needs_tools(g):
            return False
        return bool(self._SIMPLE_CONV_RE.match(g))