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 ─────────────────────────
class _AdaptiveTimeoutTracker:
    """Tracked P90 per-tool timeout con sliding window di 5 call."""
    _WINDOW     = 5
    _MIN        = 4.0   # mai sotto 4s
    _MAX        = 55.0  # mai sopra 55s
    _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:
        times = self._times.get(tool_name)
        if not times or len(times) < 2:
            return base_timeout
        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()


def _get_session_id() -> str:
    try:
        from tools.registry import _agent_session_id_var
        return _agent_session_id_var.get()
    except Exception as e:
        _logger.debug("[executor] _get_session_id failed: %s", e)
        return "default"


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
        self._circuit_recovery_counts: dict[str, int] = {}

    @classmethod
    def from_ollama(cls, ollama=None, memory=None, max_retries: int = 2) -> "Executor":
        return cls(memory=memory, max_retries=max_retries)

    def _is_circuit_open(self, tool_name: str, session_id: str) -> bool:
        tool = TOOL_REGISTRY.get(tool_name, {})
        if not tool.get("fallbacks"):
            return False
        try:
            from agents.skill_tracker import get_skill_tracker
            stats = get_skill_tracker().get_stats(session_id).get(tool_name)
        except Exception as e:
            _logger.debug("[executor] _is_circuit_open failed to get skill tracker: %s", e)
            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
        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
        return True

    async def _try_fallbacks(
        self,
        primary_name: str,
        inputs:       dict,
        timeout:      float,
        session_id:   str,
    ) -> "dict | None":
        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 as e:
            _logger.debug("[executor] _try_fallbacks failed to sort: %s", e)
            sorted_fbs = fallbacks

        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", 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)
                try:
                    from agents.skill_tracker import get_skill_tracker
                    get_skill_tracker().record(session_id, fb_name, True)
                except Exception as e:
                    _logger.debug("[executor] record success failed: %s", e)
                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 e:
                    _logger.debug("[executor] record timeout failed: %s", e)
            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 e:
                    _logger.debug("[executor] record error failed: %s", e)

        return None

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

        if self._is_circuit_open(tool_name, session_id):
            _logger.info("[executor] circuit OPEN per %s — routing a fallback", tool_name)
            fb_result = await self._try_fallbacks(tool_name, inputs, timeout, session_id)
            if fb_result:
                return fb_result
            _logger.warning("[executor] tutti i fallback di %s falliti — provo primario", tool_name)

        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:
                _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:
                    try:
                        await self.memory.save_episode("tool", f"{tool_name}: {str(inputs)[:500]}", str(result)[:500], True)
                    except Exception as e:
                        _logger.debug("[executor] memory save failed: %s", e)
                return {"success": True, "tool": tool_name, "output": result, "attempt": attempt + 1}

            except asyncio.TimeoutError:
                _timeout_tracker.record(tool_name, timeout * 1.2)
                last_error = f"Timeout dopo {timeout}s (tentativo {attempt + 1})"
                if attempt == self.max_retries:
                    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:
                    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}