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 models.ai_client import AIClient
from memory.manager import MemoryManager
from tools.registry import TOOL_REGISTRY

_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()


# ─── 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,
    ):
        self.llm         = llm_client or AIClient()
        self.memory      = memory
        self.max_retries = max_retries
        # 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)

    # ── 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:
                    pass
                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:
                    pass
            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:
                    pass

        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}

        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]  # S589: 200→300→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}