tools/harness_gate.py

"""
backend/tools/harness_gate.py — P-HARNESS: Tool Failure Tracking Gate

Traccia fallimenti per-tool per-task in memoria. Quando un tool fallisce
>= FAIL_THRESHOLD volte sullo stesso task, segnala che il provider
dovrebbe essere cambiato.

Design:
- In-memory — cleared on task end o dopo TTL (anti memory-leak)
- Fail-open: qualsiasi eccezione interna viene silenziata, mai propaga
- Thread-safe: asyncio e single-threaded, nessuna lock necessaria
- Zero dipendenze esterne aggiuntive (stdlib only)
"""
from __future__ import annotations

import logging
import re as _re
import time

_logger = logging.getLogger("tools.harness_gate")

# Soglia: N fallimenti dello stesso tool sullo stesso task -> switch provider
FAIL_THRESHOLD: int = 3

# TTL per le entry in-memory (6 ore) — evita memory leak su task lunghissimi
_ENTRY_TTL_S: float = 6.0 * 3600.0

# Struttura: task_id -> tool_name -> (fail_count, last_updated_monotonic)
_state: dict[str, dict[str, tuple[int, float]]] = {}

# Pattern: matcha "[get_weather: errore —", "[web_search: timeout", ecc.
_ERR_PREFIX_RE = _re.compile(r"^\[([a-z_]+): (?:errore|timeout|error)", _re.IGNORECASE)


# ─── API pubblica ─────────────────────────────────────────────────────────────

def record_failures_from_results(task_id: str, results: list[str]) -> int:
    """
    Analizza la lista di risultati tool e registra i fallimenti per-tool.
    Ritorna il numero di tool che hanno raggiunto FAIL_THRESHOLD.

    Chiamato alla fine di _run_direct_tools() in unified_loop_tools.py.
    Fail-open: qualsiasi eccezione interna NON propaga mai al chiamante.
    """
    try:
        _evict_stale()
        _ensure_task(task_id)
        triggered = 0
        for r in results:
            m = _ERR_PREFIX_RE.match(r)
            if not m:
                continue  # successo — non tocchiamo il contatore (tool non identificabile)
            tool_name = m.group(1).lower()
            prev_count, _ = _state[task_id].get(tool_name, (0, 0.0))
            new_count = prev_count + 1
            _state[task_id][tool_name] = (new_count, time.monotonic())
            if new_count >= FAIL_THRESHOLD:
                triggered += 1
                _logger.warning(
                    "[harness_gate] task=%s tool=%s fail_count=%d — THRESHOLD HIT, provider switch recommended",
                    task_id,
                    tool_name,
                    new_count,
                )
        return triggered
    except Exception as exc:  # noqa: BLE001
        _logger.debug("[harness_gate] silenced: %s", exc)
        return 0


def should_switch_provider(task_id: str) -> bool:
    """True se almeno un tool ha raggiunto FAIL_THRESHOLD nel task corrente."""
    try:
        if task_id not in _state:
            return False
        return any(count >= FAIL_THRESHOLD for count, _ in _state[task_id].values())
    except Exception:  # noqa: BLE001
        return False


def reset_task(task_id: str) -> None:
    """Cancella il tracking per questo task. Chiamare nel finally del loop."""
    try:
        _state.pop(task_id, None)
    except Exception:  # noqa: BLE001
        pass


def get_summary(task_id: str) -> dict[str, int]:
    """Ritorna {tool_name: fail_count} per logging/debug. Mai lancia eccezioni."""
    try:
        if task_id not in _state:
            return {}
        return {name: count for name, (count, _) in _state[task_id].items()}
    except Exception:  # noqa: BLE001
        return {}


# ─── Helpers privati ──────────────────────────────────────────────────────────

def _ensure_task(task_id: str) -> None:
    if task_id not in _state:
        _state[task_id] = {}


def _evict_stale() -> None:
    """Rimuove task entry piu vecchie di _ENTRY_TTL_S per prevenire memory leak."""
    now = time.monotonic()
    stale = [
        tid for tid, tools in _state.items()
        if not tools or all((now - ts) > _ENTRY_TTL_S for _, ts in tools.values())
    ]
    for tid in stale:
        _state.pop(tid, None)