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