Spaces:
Running
Running
File size: 4,240 Bytes
28a08e7 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 | """
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)
|