agents/executor.py

"""
executor.py — Tool Executor con retry, adaptive timeout, circuit breaker e fallback routing.
Usa AIClient (multi-provider) al posto di OllamaClient (localhost).

Architettura adaptive (GAP-SKILL-SYNC v2):
  _AdaptiveTimeoutTracker — P90-based timeout adaptation (sliding window 5 call)
  Circuit Breaker         — Wilson score < CIRCUIT_OPEN_THRESHOLD → skip al miglior fallback
  Fallback Execution      — TOOL_REGISTRY["fallbacks"] ora eseguiti automaticamente (non solo metadata)
  Recovery Credit         — tool circuit-broken retentato ogni RECOVERY_INTERVAL chiamate
"""
import asyncio
import collections
import logging
import time as _time_mod
from typing import Any

from models.ai_client import AIClient
from memory.manager import MemoryManager
from tools.registry import TOOL_REGISTRY

# P17-B1: pre-esecuzione syntax check — fail-open se ast_check non disponibile
try:
    from tools.ast_check import check_code_syntax as _check_syntax
    _CHECK_SYNTAX_AVAILABLE = True
except ImportError:
    _CHECK_SYNTAX_AVAILABLE = False
    def _check_syntax(code: str, lang: str):  # type: ignore[misc]
        class _Ok:
            ok    = True
            error = None
            line  = None
            col   = None
        return _Ok()

_logger = logging.getLogger("agente_ai.executor")

# ─── Costanti circuit breaker ────────────────────────────────────────────────
_CIRCUIT_OPEN_THRESHOLD = 0.15   # Wilson score < soglia AND >= min calls → circuit open
_MIN_CALLS_FOR_CIRCUIT  = 3      # minimo di chiamate prima che il circuit possa aprirsi
_RECOVERY_INTERVAL      = 5      # ogni N chiamate con circuit open → tenta il tool primario

# ─── S-ORCH-8GAP FIX-GAP2: Adaptive Timeout Tracker ─────────────────────────
# Sliding window (last 5 durations) per tool — calcola P90 adattivo.
# Strategia iPhone: rete variabile → se tool è stato lento di recente,
# aumenta timeout; se è stato veloce, non sprecare tempo.
class _AdaptiveTimeoutTracker:
    """Tracked P90 per-tool timeout con sliding window di 5 call."""
    _WINDOW     = 5
    _MIN        = 4.0   # mai sotto 4s — tool veloci non vanno sotto
    _MAX        = 55.0  # mai sopra 55s — iPhone connection timeout ~60s
    _MULTIPLIER = 1.5   # P90 * 1.5 = headroom conservativo

    def __init__(self) -> None:
        self._times: dict[str, collections.deque] = {}

    def record(self, tool_name: str, elapsed: float) -> None:
        if tool_name not in self._times:
            self._times[tool_name] = collections.deque(maxlen=self._WINDOW)
        self._times[tool_name].append(elapsed)

    def adaptive_timeout(self, tool_name: str, base_timeout: float) -> float:
        """Ritorna timeout adattivo: P90 * 1.5 se dati sufficienti, else base."""
        times = self._times.get(tool_name)
        if not times or len(times) < 2:
            return base_timeout  # dati insufficienti → usa base invariato
        sorted_t = sorted(times)
        p90_idx  = min(int(len(sorted_t) * 0.9), len(sorted_t) - 1)
        adaptive = sorted_t[p90_idx] * self._MULTIPLIER
        return max(self._MIN, min(self._MAX, adaptive))

_timeout_tracker = _AdaptiveTimeoutTracker()

# P17-B1: mapping tool_name → (argomento_codice, linguaggio) per syntax check
_CODE_EXEC_TOOLS: dict[str, tuple[str, str]] = {
    "run_python": ("code", "python"),
    "run_code":   ("code", "python"),
}


# ─── Helper: ottieni session_id dal ContextVar (impostato da unified_loop.py) ─
def _get_session_id() -> str:
    try:
        from tools.registry import _agent_session_id_var
        return _agent_session_id_var.get()
    except Exception:
        return "default"


# ─── Executor ────────────────────────────────────────────────────────────────

class Executor:
    def __init__(
        self,
        llm_client:  AIClient | None = None,
        memory:      MemoryManager | None = None,
        max_retries: int = 2,
        kernel:      Any | None = None,   # ARCH-K2.2: Brain→Kernel abstraction
    ):
        self.llm         = llm_client or AIClient()
        self.memory      = memory
        self.max_retries = max_retries
        self._kernel     = kernel         # ARCH-K2.2: usato da submit_background_task()
        # GAP-SKILL-SYNC v2: contatore chiamate per recovery credit (per-tool)
        self._circuit_recovery_counts: dict[str, int] = {}

    # Backward-compat: vecchia firma aveva ollama=OllamaClient, memory=MemoryManager
    @classmethod
    def from_ollama(cls, ollama=None, memory=None, max_retries: int = 2) -> "Executor":
        return cls(memory=memory, max_retries=max_retries)

    # ── ARCH-K2.2: submit background task via Kernel ──────────────────────────

    async def submit_background_task(
        self,
        payload:    dict,
        priority:   str = "BACKGROUND",
        session_id: str | None = None,
    ) -> str | None:
        """
        Invia un task in background tramite kernel.submit_task() (ARCH-K2.2).

        Il Brain/Executor non conosce l'implementazione della coda sottostante
        (S9: ogni servizio ignora l'impl interna degli altri).

        Fallback: asyncio.create_task() locale se il Kernel non è disponibile.
        Sempre non-bloccante — non aspetta il completamento del task.

        Ritorna il task_id se il Kernel è disponibile, None altrimenti.
        """
        # Lazy-load kernel singleton se non iniettato
        k = self._kernel
        if k is None:
            try:
                from api.kernel import kernel as _k
                k = _k
            except Exception:
                pass

        if k is not None:
            try:
                result = await k.submit_task(
                    payload=payload,
                    priority=priority,
                    session_id=session_id,
                )
                _logger.info(
                    "[executor] submit_background_task via Kernel id=%s priority=%s",
                    result.task_id, priority,
                )
                return result.task_id
            except Exception as exc:
                _logger.warning("[executor] kernel submit_background_task err: %s", exc)

        # Fallback: esecuzione diretta asincrona locale (non attraverso la Queue)
        _logger.debug("[executor] submit_background_task fallback: asyncio.create_task")
        return None

    # ── Circuit breaker helper ────────────────────────────────────────────────

    def _is_circuit_open(self, tool_name: str, session_id: str) -> bool:
        """True se il circuit breaker deve aprirsi per questo tool in questa sessione.

        Condizioni (tutte necessarie):
          1. Wilson score < CIRCUIT_OPEN_THRESHOLD (0.15)
          2. >= MIN_CALLS_FOR_CIRCUIT (3) chiamate nella sessione
          3. Il tool ha fallback disponibili in TOOL_REGISTRY
        Recovery credit: ogni RECOVERY_INTERVAL chiamate, il circuit si chiude
        temporaneamente per un tentativo di recovery.
        """
        tool = TOOL_REGISTRY.get(tool_name, {})
        if not tool.get("fallbacks"):
            return False  # senza fallback il circuit non può aprirsi
        try:
            from agents.skill_tracker import get_skill_tracker
            stats = get_skill_tracker().get_stats(session_id).get(tool_name)
        except Exception:
            return False
        if not stats:
            return False
        if stats["total_count"] < _MIN_CALLS_FOR_CIRCUIT:
            return False
        if stats["wilson_score"] >= _CIRCUIT_OPEN_THRESHOLD:
            return False
        # Recovery credit: conta le chiamate e apri una finestra ogni RECOVERY_INTERVAL
        count = self._circuit_recovery_counts.get(tool_name, 0) + 1
        self._circuit_recovery_counts[tool_name] = count
        if count % _RECOVERY_INTERVAL == 0:
            _logger.info(
                "[executor] recovery credit: riprovo %s (circuit call #%d)",
                tool_name, count,
            )
            return False   # consenti un tentativo di recovery
        return True

    # ── Fallback execution ────────────────────────────────────────────────────

    async def _try_fallbacks(
        self,
        primary_name: str,
        inputs:       dict,
        timeout:      float,
        session_id:   str,
    ) -> "dict | None":
        """Tenta i fallback definiti in TOOL_REGISTRY ordinati per Wilson score.

        Registra ogni tentativo nel skill_tracker sotto il nome del fallback.
        Ritorna il primo risultato con successo, o None se tutti falliscono.
        """
        tool      = TOOL_REGISTRY.get(primary_name, {})
        fallbacks = tool.get("fallbacks", [])
        if not fallbacks:
            return None

        try:
            from agents.skill_tracker import get_skill_tracker
            sorted_fbs = get_skill_tracker().get_sorted_fallbacks(session_id, fallbacks)
        except Exception:
            sorted_fbs = fallbacks   # ordinamento originale come fallback del fallback

        for fb_name in sorted_fbs:
            fb_tool = TOOL_REGISTRY.get(fb_name)
            if not fb_tool or not fb_tool.get("_fn"):
                continue
            _logger.info(
                "[executor] %s fallita — provo fallback %s (Wilson-sorted)",
                primary_name, fb_name,
            )
            try:
                _t0     = _time_mod.monotonic()
                _fb_to  = _timeout_tracker.adaptive_timeout(fb_name, timeout)
                result  = await asyncio.wait_for(fb_tool["_fn"](**inputs), timeout=_fb_to)
                _timeout_tracker.record(fb_name, _time_mod.monotonic() - _t0)
                # Registra il successo del fallback nel skill_tracker
                try:
                    from agents.skill_tracker import get_skill_tracker
                    get_skill_tracker().record(session_id, fb_name, True)
                except Exception as _skt_err:
                    _logger.debug("[executor] skill_tracker silenced: %s", _skt_err)  # BUG-SILENT-EXC
                return {
                    "success":            True,
                    "tool":               fb_name,
                    "output":             result,
                    "via_fallback_from":  primary_name,
                    "attempt":            1,
                }
            except asyncio.TimeoutError:
                _timeout_tracker.record(fb_name, timeout * 1.2)
                _logger.debug("[executor] fallback %s timeout", fb_name)
                try:
                    from agents.skill_tracker import get_skill_tracker
                    get_skill_tracker().record(session_id, fb_name, False)
                except Exception as _skt_err:
                    _logger.debug("[executor] skill_tracker silenced: %s", _skt_err)  # BUG-SILENT-EXC
            except Exception as fb_exc:
                _logger.debug("[executor] fallback %s errore: %s", fb_name, str(fb_exc)[:80])
                try:
                    from agents.skill_tracker import get_skill_tracker
                    get_skill_tracker().record(session_id, fb_name, False)
                except Exception as _skt_err:
                    _logger.debug("[executor] skill_tracker silenced: %s", _skt_err)  # BUG-SILENT-EXC

        return None   # tutti i fallback hanno fallito

    # ── run_tool ─────────────────────────────────────────────────────────────

    async def run_tool(self, tool_name: str, inputs: dict, timeout: float = 30.0) -> dict:
        tool = TOOL_REGISTRY.get(tool_name)
        if not tool:
            return {"success": False, "error": f"Tool '{tool_name}' non trovato", "output": None}

        missing = [r for r in tool.get("required_inputs", []) if r not in inputs]
        if missing:
            return {"success": False, "error": f"Input mancanti: {missing}", "output": None}

        session_id = _get_session_id()

        # ── GAP-SKILL-SYNC v2: circuit breaker pre-check ──────────────────────
        # Se il tool ha un Wilson score molto basso (< 0.15) con >= 3 dati in sessione,
        # bypassa il tool e vai direttamente al miglior fallback disponibile.
        if self._is_circuit_open(tool_name, session_id):
            _logger.info(
                "[executor] circuit OPEN per %s — routing diretto a fallback (Wilson < %.2f)",
                tool_name, _CIRCUIT_OPEN_THRESHOLD,
            )
            fb_result = await self._try_fallbacks(tool_name, inputs, timeout, session_id)
            if fb_result:
                return fb_result
            # Tutti i fallback falliti: procedi con il tool primario (ultima spiaggia)
            _logger.warning(
                "[executor] tutti i fallback di %s hanno fallito — provo comunque il tool primario",
                tool_name,
            )

        # ── Esecuzione normale con retry ──────────────────────────────────────
        fn = tool.get("_fn")
        if fn is None:
            return {"success": False, "error": "Tool non ha funzione di esecuzione", "output": None}

        # P17-B1: syntax check pre-esecuzione — intercetta SyntaxError prima che il
        # backend-exec spreci un round-trip su codice già rotto. Fail-open: tool non in
        # mappa, ast_check non importato, o codice vuoto → nessun blocco.
        if tool_name in _CODE_EXEC_TOOLS:
            _code_arg, _code_lang = _CODE_EXEC_TOOLS[tool_name]
            _raw_code = inputs.get(_code_arg, "")
            if isinstance(_raw_code, str) and _raw_code.strip():
                _syn = _check_syntax(_raw_code, _code_lang)
                if not _syn.ok:
                    _logger.warning(
                        "[executor] P17-B1 syntax check failed per %s: %s",
                        tool_name, _syn.error,
                    )
                    return {
                        "success": False,
                        "error": (
                            f"SyntaxError pre-esecuzione [{_code_lang}]: {_syn.error}"
                            + (f" — riga {_syn.line}" if _syn.line else "")
                        ),
                        "output": None,
                        "syntax_check_failed": True,
                    }

        last_error: str = "max_retries"
        for attempt in range(self.max_retries + 1):
            try:
                # S-ORCH-8GAP FIX-GAP2: usa timeout adattivo basato su P90 ultime 5 chiamate
                _adaptive_to = _timeout_tracker.adaptive_timeout(tool_name, timeout)
                _t0          = _time_mod.monotonic()
                result       = await asyncio.wait_for(fn(**inputs), timeout=_adaptive_to)
                _timeout_tracker.record(tool_name, _time_mod.monotonic() - _t0)
                if self.memory:
                    # S577→S600: inputs 100→500 — parity con altri handler
                    await self.memory.save_episode(
                        "tool",
                        f"{tool_name}: {str(inputs)[:500]}",
                        str(result)[:500],
                        True,
                    )
                return {"success": True, "tool": tool_name, "output": result, "attempt": attempt + 1}

            except asyncio.TimeoutError:
                # FIX-GAP2: registra il timeout come durata massima per shrink futuro
                _timeout_tracker.record(tool_name, timeout * 1.2)
                last_error = f"Timeout dopo {timeout}s (tentativo {attempt + 1})"
                if attempt == self.max_retries:
                    # Ultima chance: prova i fallback ordinati per Wilson score
                    _logger.info(
                        "[executor] %s timeout definitivo — provo fallback Wilson-sorted",
                        tool_name,
                    )
                    fb_result = await self._try_fallbacks(tool_name, inputs, timeout, session_id)
                    if fb_result:
                        return fb_result
                    return {"success": False, "error": last_error, "output": None}
                await asyncio.sleep(0.5)

            except Exception as e:
                last_error = str(e)
                if attempt == self.max_retries:
                    # Ultima chance: prova i fallback ordinati per Wilson score
                    _logger.info(
                        "[executor] %s errore definitivo (%s) — provo fallback Wilson-sorted",
                        tool_name, last_error[:60],
                    )
                    fb_result = await self._try_fallbacks(tool_name, inputs, timeout, session_id)
                    if fb_result:
                        return fb_result
                    return {"success": False, "error": last_error, "output": None}
                await asyncio.sleep(0.5)

        return {"success": False, "error": f"Max retries raggiunti: {last_error}", "output": None}