api/_agent_helpers.py

"""backend/api/_agent_helpers.py — Funzioni helper condivise tra i sub-router agent.

Estratto da agent_loop_routes / agent_task_routes / agent_checkpoint_routes
per eliminare le triplicazioni create dallo split S359 (2026-06-30).

Esportazioni:
  _RE_SURROGATES         — regex surrogati UTF-16
  _ss(s)                 — sanitizza string (rimuove surrogati)
  _log_task_exc(task)    — done-callback asyncio con log eccezioni
  _PERSONA_KEYWORD_MAP   — dict globale (regex per persona routing)
  _PERSONA_CLIENT_CACHE  — dict globale (cache LLM client per persona)
  _build_persona_kw_map  — costruisce _PERSONA_KEYWORD_MAP (P17-F5)
  _classify_persona_server — classifica persona via regex scoring (zero LLM)
  _get_persona_llm_client  — ritorna LLM client persona-appropriato
"""
from __future__ import annotations
import re
import logging

_logger = logging.getLogger("api.agent")

_RE_SURROGATES = re.compile(r"[\uD800-\uDFFF]", re.UNICODE)
def _ss(s: object) -> str:
    if not isinstance(s, str):
        return s
    try:
        cleaned = _RE_SURROGATES.sub("", s)
        cleaned = cleaned.encode("utf-8", errors="replace").decode("utf-8", errors="replace")
    except Exception:
        cleaned = s
    return cleaned
def _log_task_exc(task):
    if not task.cancelled():
        exc = task.exception()
        if exc:
            _logger.warning("[agent] background task raised %s: %s", type(exc).__name__, exc)
try:
    from .telegram_notify import notify_task_done as _tg_done, notify_task_error as _tg_error, notify_task_start as _tg_start, notify_task_step as _tg_step
except Exception:
    async def _tg_done(*_a, **_kw): pass  # type: ignore[misc]
    async def _tg_error(*_a, **_kw): pass  # type: ignore[misc]
    async def _tg_start(*_a, **_kw): pass  # type: ignore[misc]
    async def _tg_step(*_a, **_kw):  pass  # type: ignore[misc]

router = APIRouter()

@router.post('/run_loop', deprecated=True)
async def run_loop():
    """Deprecated — use /agent/task instead."""
    from fastapi.responses import JSONResponse
    return JSONResponse(status_code=410, content={"detail": {"error": "Gone", "migration": "/api/agent/tasks"}})


# ─── P17-F5: Persona helpers ──────────────────────────────────────────────────
import re as _re_persona

_PERSONA_KEYWORD_MAP: dict = {}

def _build_persona_kw_map() -> dict:
    import re
    return {
        'researcher': re.compile(
            r'\b(cerca|ricerca|research|trova|notizie|news|url|leggi|articolo|wikipedia|'
            r'google|fonte|source|scrape|fetch|sito|pagina|web|http|verifica|fact.?check)\b',
            re.IGNORECASE
        ),
        'coder': re.compile(
            r'\b(codice|code|funzione|function|bug|script|implementa|python|javascript|'
            r'typescript|refactor|debug|test|classe|class|api|endpoint|sql|database|html|'
            r'css|react|app|applicazione|programma|sviluppa)\b',
            re.IGNORECASE
        ),
        'reasoner': re.compile(
            r'\b(analizza|pianifica|strategia|decide|ragiona|valuta|confronta|'
            r'piano|roadmap|architettura|valutazione|decisione|ottimale|consiglia)\b',
            re.IGNORECASE
        ),
        'analyst': re.compile(
            r'\b(dati|statistiche|grafico|dataset|csv|dataframe|pandas|matplotlib|'
            r'metriche|kpi|trend|visualizza|dashboard|excel|tabella|percentuale|distribuzione)\b',
            re.IGNORECASE
        ),
    }

def _classify_persona_server(goal: str) -> str:
    """P17-F5: classifica la persona dal goal via regex scoring. Zero LLM — zero latency."""
    global _PERSONA_KEYWORD_MAP
    if not _PERSONA_KEYWORD_MAP:
        _PERSONA_KEYWORD_MAP = _build_persona_kw_map()
    if not goal or len(goal) < 4:
        return ''
    best, best_score = '', 0
    for persona_id, pattern in _PERSONA_KEYWORD_MAP.items():
        score = len(pattern.findall(goal))
        if score > best_score:
            best_score, best = score, persona_id
    return best if best_score >= 1 else ''

_PERSONA_CLIENT_CACHE: dict = {}

def _get_persona_llm_client(persona: str, default_client: object) -> object:
    """P17-F5: ritorna il client LLM persona-appropriate via role_router.
    Fallback silente su default_client se la chiave API manca o role_router fallisce.
    Cache in-process — zero overhead dopo il primo accesso.""";
    if not persona:
        return default_client
    if persona in _PERSONA_CLIENT_CACHE:
        return _PERSONA_CLIENT_CACHE[persona]
    _ROLE_MAP = {'researcher': 'RESEARCHER', 'analyst': 'RESEARCHER',
                 'coder': 'CODER', 'reasoner': 'REASONER', 'architect': 'ARCHITECT'}
    role_name = _ROLE_MAP.get(persona.lower())
    if not role_name:
        return default_client
    try:
        from models.role_router import RoleRouter, Role as _Role
        role = getattr(_Role, role_name, None)
        if role is None:
            return default_client
        client = RoleRouter.get_client(role)
        _PERSONA_CLIENT_CACHE[persona] = client
        return client
    except Exception:
        return default_client