"""backend/api/providers.py — Health, tools, status, AI health, heartbeat (S354).""" import os, asyncio, time, logging from fastapi import APIRouter, Request from .state import _sb, SENSITIVE, _ai_health_cache, _AI_HEALTH_TTL, _heartbeat_state, _TIMING_STORE, _REPAIR_STATS router = APIRouter() _logger = logging.getLogger('agente_ai') _BOOT_TIME = time.monotonic() # S-DUAL-1: uptime per /api/health/load # S388: intervallo ridotto 300→90s — provider down rilevati in ≤90s invece di 5min. # Configurabile via env HEARTBEAT_INTERVAL per deployment che vogliono più o meno frequenza. _HEARTBEAT_INTERVAL_S = int(os.getenv("HEARTBEAT_INTERVAL", "90")) _HEARTBEAT_WARMUP_TOKENS = 32 # era 3 — alzato per evitare 400 BadRequest su provider con min_tokens (SambaNova, CF, Groq) # S442-FIX1: singleton guard — previene task duplicati se start_heartbeat() chiamata 2+ volte. # Scenario: riavvio anomalo del lifespan, hot-reload, test runner che importa più volte. # _heartbeat_task è None prima del primo avvio, poi punta all'asyncio.Task in corso. # Se il task è done() (crash/cancel), viene riavviato. _heartbeat_task: asyncio.Task | None = None # ── Health / Status ──────────────────────────────────────────────────────────── @router.get('/health') @router.get('/api/health') # alias — Railway health checks usano /api/health async def health(): return { 'status': 'ok', 'version': '3.4.2', 'supabase': _sb is not None, 'backend': 'HuggingFace Spaces / Railway', } # ── S-DUAL-1: Load metrics for CF dual-space adaptive routing ───────────────── @router.get('/api/health/load') async def health_load(): """ Metriche di carico per il routing adattivo del CF Pages Function (S-DUAL-1). Usato dal router CF per decidere se HANDS è saturo prima di fare fallback. Non richiede auth — dati aggregati, nessun dato sensibile. Campi risposta: space_role "brain" | "hands" | "unknown" (env SPACE_ROLE) active_agent_tasks task agent in stato RUNNING in questa istanza realtime_active job exec/shell correnti (semaphore REALTIME) realtime_capacity max job REALTIME concorrenti realtime_available slot REALTIME liberi background_active job benchmark/research/pip correnti background_capacity max job BACKGROUND concorrenti background_available slot BACKGROUND liberi uptime_s secondi dall'avvio del processo uvicorn ts timestamp ms """ from .state import _agent_tasks try: from .priority import get_load_metrics as _glm _metrics = _glm() except Exception: _metrics = { "realtime_active": 0, "realtime_capacity": 6, "realtime_available": 6, "background_active": 0, "background_capacity": 2, "background_available": 2, "uptime_s": int(time.monotonic() - _BOOT_TIME), } _active_tasks = sum( 1 for t in _agent_tasks.values() if t.get('status') in ('RUNNING', 'running') ) return { 'space_role': os.getenv('SPACE_ROLE', 'unknown'), 'active_agent_tasks': _active_tasks, 'realtime_active': _metrics['realtime_active'], 'realtime_capacity': _metrics['realtime_capacity'], 'realtime_available': _metrics['realtime_available'], 'background_active': _metrics['background_active'], 'background_capacity': _metrics['background_capacity'], 'background_available': _metrics['background_available'], 'uptime_s': _metrics['uptime_s'], 'ts': int(time.time() * 1000), } @router.get('/api/version') async def api_version(): """S456-X3: versione dettagliata con sprint, capabilities e soglie refusal. Il frontend legge questo endpoint all'avvio per verificare l'allineamento tra la versione del loop browser (agentLoop.ts) e il loop backend (unified_loop.py). """ return { 'sprint': 'S766-RC1', 'version': '3.5.0', 'build_date': '2026-06-19', 'capabilities': [ 'never_give_up', # S197: retry forzato su rifiuto LLM 'reflective_debug', # S455-P14: fallback chain _reflective_debug 'structured_memory', # S401: projectMemory con sessionStorage backup 'goal_verifier_v2', # S410: GoalVerifier 2.0 con coverage check 'speculative_tools', # S361: pre-fire tool speculativi in parallelo 'project_context', # S456-X5: project memory iniettato dal frontend 'learning_hints', # S456-X4: failure pattern dal selfLearning frontend 'severity_retry', # S376: retry adattivo syntax/runtime/logic 'consensus_mode', # S91: multi-provider consensus su task complessi 'vision_tools', # V001: analyze_image/generate_image/search_images/screenshot 'email_send', # V002: send_email via Resend API 'database_query', # V003: PostgreSQL + SQLite query 'web_research', # V004: multi-URL research + Groq synthesis 'execute_sql', # V005: SQL execution frontend sandbox 'create_pdf', # V006: PDF generation frontend (jsPDF) 'call_api', # V007: direct REST API calls 'graph_orchestrator', # S760: GraphOrchestrator parallel node execution + S760-A/B/C/D resilience 'jit_planning', # S-JIT: Just-In-Time planner (800ms timeout, 0ms local fallback) 'sched_sse', # S-SCHED-SSE: Scheduler SSE real-time push (<100ms latency) 'lru_cache_n', # S766: LRU-N selfLearningWorker (LRU-3 context, LRU-5 experience) 'nvidia_nim', # NVIDIA NIM provider — 15 modelli verificati (integrate.api.nvidia.com) 'role_fast', # S-FAST: Role.FAST path — Groq 8B per query semplici (<200ms) 'bg_task_recovery', # S-PERSIST: task persistenti + BgTaskRecoveryBanner ], 'refusal': { # Soglie INTENZIONALMENTE separate (azioni diverse): # threshold_retry: backend retry aggressivo (cheap) → soglia alta # threshold_validate: frontend quality penalty (conservativo) → soglia bassa 'threshold_retry': 600, 'threshold_validate': 350, 'phrases_canonical': True, # S456-X2: set di frasi sincronizzato frontend/backend }, } @router.get('/api/tools') async def list_tools(): try: from tools.registry import TOOL_REGISTRY tools_list = [ { "name": spec["name"], "description": spec.get("description", spec.get("goal", spec["name"])), "required_inputs": spec.get("required_inputs", []), "optional_inputs": spec.get("optional_inputs", {}), "risk_level": spec.get("risk_level", "unknown"), } for spec in TOOL_REGISTRY.values() ] return {"tools": tools_list, "count": len(tools_list), "status": "ok"} except Exception as exc: return {"tools": [], "count": 0, "error": str(exc)} @router.get('/api/status') async def status(request: Request): # security-fix: richiede X-Internal-Token — endpoint espone env vars _tok = os.getenv('INTERNAL_TOKEN', '') if _tok and request.headers.get('X-Internal-Token', '') != _tok: from fastapi import HTTPException as _HTTPEx raise _HTTPEx(401, 'Unauthorized') safe_env = {k: '***' if k in SENSITIVE else v for k, v in os.environ.items()} return {'status': 'running', 'env': safe_env, 'supabase': _sb is not None} @router.get('/api/ai/health') async def ai_provider_health(): """Testa tutti i provider AI in parallelo — risultati cachati 60s.""" now = time.monotonic() if _ai_health_cache["data"] and now - _ai_health_cache["at"] < _AI_HEALTH_TTL: return _ai_health_cache["data"] from models.ai_client import AIClient client = AIClient() async def _probe(provider) -> dict: t0 = time.monotonic() try: c = client._client_for(provider) await asyncio.wait_for( asyncio.to_thread( c.chat.completions.create, model=provider.default_model, messages=[{"role": "user", "content": "1+1="}], max_tokens=5, stream=False, ), timeout=8.0, ) ms = round((time.monotonic() - t0) * 1000) return {"name": provider.name, "ok": True, "status": "ok", "latency_ms": ms, "model": provider.default_model.split("/")[-1][:28]} except Exception as exc: ms = round((time.monotonic() - t0) * 1000) return {"name": provider.name, "ok": False, "status": "error", "latency_ms": ms, "error": str(exc)[:300], "model": provider.default_model.split("/")[-1][:28]} # S606: 200→300 results = list(await asyncio.gather(*[_probe(p) for p in client.providers])) payload = {"providers": results, "tested_at": int(time.time() * 1000)} _ai_health_cache["data"] = payload _ai_health_cache["at"] = time.monotonic() return payload # ── GAP-PROVIDER-FIX: canonical provider order ────────────────────────────────── @router.get("/api/providers/canonical") async def providers_canonical(): """ GAP-PROVIDER-FIX: espone l'ordine di priorità backend dei provider in modo leggibile dal frontend (providerBridge) — elimina la divergenza silenziosa tra routing frontend (intent-based, Gemini-first) e backend (Groq-first sequential). Ritorna la lista live da AIClient._discover_providers() nell'ordine esatto di fallback usato da ai_client.chat(). """ try: from models.ai_client import AIClient client = AIClient() return { "providers": [ { "name": p.name, "model": p.default_model.split("/")[-1][:40], "priority": i, } for i, p in enumerate(client.providers) ], "count": len(client.providers), "primary": client.providers[0].name if client.providers else None, "note": ( "Backend usa sequential fallback (groq→cerebras→sambanova→gemini…). " "Frontend usa intent-based routing per-task. Ordini divergono per design." ), } except Exception as exc: return {"providers": [], "count": 0, "error": str(exc)[:200]} # ── Provider heartbeat ───────────────────────────────────────────────────────── async def _heartbeat_probe_all() -> list: try: from models.ai_client import AIClient client = AIClient() async def _probe(provider) -> dict: t0 = time.monotonic() try: c = client._client_for(provider) await asyncio.wait_for( asyncio.to_thread( c.chat.completions.create, model=provider.default_model, messages=[{"role": "user", "content": "ok"}], max_tokens=_HEARTBEAT_WARMUP_TOKENS, stream=False, ), timeout=10.0, ) ms = round((time.monotonic() - t0) * 1000) return {"name": provider.name, "ok": True, "latency_ms": ms} except Exception as exc: ms = round((time.monotonic() - t0) * 1000) return {"name": provider.name, "ok": False, "latency_ms": ms, "error": str(exc)[:300]} # S606: 200→300 return list(await asyncio.gather(*[_probe(p) for p in client.providers])) except Exception as exc: _logger.warning("heartbeat probe failed: %s", exc) return [] async def _heartbeat_loop() -> None: # S388: warmup delay ridotto 10s→2s — heartbeat inizia quasi subito dopo il boot. await asyncio.sleep(2) while True: _heartbeat_state["status"] = "running" _heartbeat_state["last_run_at"] = int(time.time()) _heartbeat_state["next_run_at"] = int(time.time()) + _HEARTBEAT_INTERVAL_S try: results = await _heartbeat_probe_all() available = [r for r in results if r.get("ok")] best = min(available, key=lambda r: r["latency_ms"]) if available else None _heartbeat_state["providers"] = results _heartbeat_state["best_provider"] = best["name"] if best else None _heartbeat_state["best_latency_ms"] = best["latency_ms"] if best else None _heartbeat_state["runs"] += 1 _heartbeat_state["status"] = "ok" _heartbeat_state["error"] = None _logger.info( "heartbeat #%d: best=%s (%dms), available=%d/%d", _heartbeat_state["runs"], _heartbeat_state["best_provider"], _heartbeat_state["best_latency_ms"] or 0, len(available), len(results), ) _ai_health_cache["data"] = {"providers": results, "tested_at": int(time.time() * 1000)} _ai_health_cache["at"] = time.monotonic() except Exception as exc: _heartbeat_state["status"] = "error" _heartbeat_state["error"] = str(exc)[:300] # S588: 200→300 _logger.error("heartbeat error: %s", exc) # MX12-P2: adaptive interval — dimezza quando <2 provider disponibili # per rilevare il recovery più velocemente senza aumentare il carico base. _available_count = len([r for r in _heartbeat_state.get("providers", []) if r.get("ok")]) _adaptive_s = max(30, _HEARTBEAT_INTERVAL_S // 2) if _available_count < 2 else _HEARTBEAT_INTERVAL_S if _adaptive_s != _HEARTBEAT_INTERVAL_S: _logger.info("heartbeat adaptive: %ds (providers ok=%d)", _adaptive_s, _available_count) await asyncio.sleep(_adaptive_s) def start_heartbeat() -> None: """Avvia il loop heartbeat — chiamato da main.py startup event. S442-FIX1: singleton guard — crea il task solo se non esiste già o se è crashed/cancelled. """ global _heartbeat_task try: loop = asyncio.get_event_loop() if not loop.is_running(): return # Guard: non avviare se il task è ancora vivo if _heartbeat_task is not None and not _heartbeat_task.done(): _logger.info("start_heartbeat: task già in esecuzione, skip duplicato") return _heartbeat_task = asyncio.create_task(_heartbeat_loop()) _logger.info("start_heartbeat: task avviato (pid=%s)", id(_heartbeat_task)) except Exception as exc: _logger.warning("start_heartbeat failed: %s", exc) # ── MX12-P1: get_best_provider_fast() — TTL-guarded in-memory read ──────────── # Usato da unified_loop e qualsiasi client interno che vuole il provider migliore # senza trigger network. Se heartbeat è stale (>_PROVIDER_CACHE_TTL_S) restituisce # il fallback configurabile via env DEFAULT_PROVIDER (default: "groq"). # Thread-safe: legge solo dict primitivi Python (GIL garantisce letture atomiche). _PROVIDER_CACHE_TTL_S = int(os.getenv("PROVIDER_CACHE_TTL", "60")) def get_best_provider_fast() -> str: """Provider migliore dalla cache in-memory senza network calls. Regola TTL: - Se heartbeat ha girato entro _PROVIDER_CACHE_TTL_S → usa best_provider - Se stale o heartbeat non ancora partito → fallback DEFAULT_PROVIDER Fallback: 'groq' (tier free 900k tok/giorno, latenza <300ms tipica) """ last = _heartbeat_state.get("last_run_at") or 0 age = int(time.time()) - last best = _heartbeat_state.get("best_provider") if best and age <= _PROVIDER_CACHE_TTL_S: return best fallback = os.getenv("DEFAULT_PROVIDER", "groq") if age > _PROVIDER_CACHE_TTL_S and last > 0: _logger.debug("get_best_provider_fast: stale (%ds) — fallback %s", age, fallback) return fallback @router.get("/debug/timing") async def debug_timing(): """S385: Latency telemetry — p50/p95/min/max per metrica LLM e tool call.""" def _pct(values: list[float], p: float) -> float | None: if not values: return None s = sorted(values) idx = int(len(s) * p) return round(s[min(idx, len(s) - 1)], 1) def _stats(label: str) -> dict: samples: list[float] = _TIMING_STORE.get(label, []) _avg = round(sum(samples) / len(samples), 1) if samples else None return { "count": len(samples), "avg": _avg, "p50_ms": _pct(samples, 0.50), "p95_ms": _pct(samples, 0.95), "min_ms": round(min(samples), 1) if samples else None, "max_ms": round(max(samples), 1) if samples else None, } _timings = { "llm_first_token": _stats("llm_first_token"), "llm_total": _stats("llm_total"), "tool_call": _stats("tool_call"), "direct_tool": _stats("direct_tool"), # Sprint 5 ITEM 13: phase breakdown — medie per fase del loop agente "classify_ms": _stats("classify_ms"), "plan_ms": _stats("plan_ms"), "coder_ms": _stats("coder_ms"), "verifier_ms": _stats("verifier_ms"), "browser_ms": _stats("browser_ms"), } return { "server_time_ms": int(time.time() * 1000), "timings": _timings, "timing_stats": _timings, # alias per retrocompatibilità frontend "repair_stats": dict(_REPAIR_STATS), "best_provider": _heartbeat_state.get("best_provider"), "best_latency_ms": _heartbeat_state.get("best_latency_ms"), } @router.get("/api/providers/heartbeat") async def providers_heartbeat(): now = int(time.time()) return { "status": _heartbeat_state["status"], "best_provider": _heartbeat_state["best_provider"], "best_latency_ms": _heartbeat_state["best_latency_ms"], "providers": _heartbeat_state["providers"], "last_run_at": _heartbeat_state["last_run_at"], "next_run_at": _heartbeat_state["next_run_at"], "runs": _heartbeat_state["runs"], "error": _heartbeat_state["error"], "interval_s": _HEARTBEAT_INTERVAL_S, "server_time": now, } # ── /api/health/full — aggregated health (AI + Supabase + Telegram + backend) ── @router.get("/api/health/full") async def health_full(): """ Aggregated health check in un'unica chiamata. Risponde in <200ms: - ai: da _heartbeat_state (cache, no LLM probe live) - supabase: probe live SELECT limit 1 (timeout 3s) - telegram: /getMe live (timeout 3s) - backend: task attivi, heartbeat runs, timing snapshot Usato da monitoring, dashboard prod e debug. """ import time as _t t0 = _t.monotonic() now = int(_t.time()) # ── 1. AI providers — heartbeat cache, istantaneo ───────────────────────── hb = _heartbeat_state _hb_providers = hb.get("providers", []) _hb_ok = [p for p in _hb_providers if p.get("ok")] _last_run = hb.get("last_run_at") or 0 ai_section = { "ok": len(_hb_ok) > 0, "available": len(_hb_ok), "total": len(_hb_providers), "best": hb.get("best_provider"), "best_latency_ms": hb.get("best_latency_ms"), "providers": _hb_providers, "last_probe_at": _last_run, "next_probe_at": hb.get("next_run_at"), "stale": (now - _last_run) > (_HEARTBEAT_INTERVAL_S * 2) if _last_run else True, "heartbeat_runs": hb.get("runs", 0), } # ── 2. Supabase — probe live (SELECT 1 via table scan, timeout 3s) ──────── async def _probe_supabase() -> dict: if _sb is None: return {"ok": False, "configured": False, "error": "not_configured", "detail": "Imposta SUPABASE_URL + SUPABASE_KEY in Railway/HF Spaces"} _pt = _t.monotonic() try: await asyncio.wait_for( asyncio.to_thread( lambda: _sb.table("agent_tasks").select("task_id").limit(1).execute() ), timeout=3.0, ) return {"ok": True, "configured": True, "latency_ms": round((_t.monotonic() - _pt) * 1000)} except asyncio.TimeoutError: return {"ok": False, "configured": True, "error": "timeout", "latency_ms": round((_t.monotonic() - _pt) * 1000)} except Exception as exc: return {"ok": False, "configured": True, "latency_ms": round((_t.monotonic() - _pt) * 1000), "error": str(exc)[:150]} # ── 3. Telegram — /getMe live (timeout 3s) ──────────────────────────────── async def _probe_telegram() -> dict: _pt = _t.monotonic() try: from .telegram_notify import _load_config as _tg_cfg cfg = await asyncio.wait_for(_tg_cfg(), timeout=2.0) if not cfg or not cfg.get("token"): return {"ok": False, "configured": False, "detail": "Imposta TELEGRAM_BOT_TOKEN + TELEGRAM_CHAT_ID in Railway"} import httpx async with httpx.AsyncClient(timeout=3.0) as hc: r = await hc.get(f"https://api.telegram.org/bot{cfg['token']}/getMe") ms = round((_t.monotonic() - _pt) * 1000) if r.status_code == 200 and r.json().get("ok"): bot = r.json()["result"] return { "ok": True, "configured": True, "latency_ms": ms, "username": bot.get("username"), "bot_id": bot.get("id"), "chat_id_set": bool(cfg.get("chat_id")), } return {"ok": False, "configured": True, "latency_ms": ms, "error": f"HTTP {r.status_code}: {r.text[:120]}"} except asyncio.TimeoutError: return {"ok": False, "configured": None, "error": "timeout", "latency_ms": round((_t.monotonic() - _pt) * 1000)} except Exception as exc: return {"ok": False, "configured": None, "error": str(exc)[:150], "latency_ms": round((_t.monotonic() - _pt) * 1000)} # ── run supabase + telegram in parallelo ────────────────────────────────── sb_res, tg_res = await asyncio.gather( _probe_supabase(), _probe_telegram(), return_exceptions=True ) if isinstance(sb_res, Exception): sb_res = {"ok": False, "error": str(sb_res)[:150]} if isinstance(tg_res, Exception): tg_res = {"ok": False, "error": str(tg_res)[:150]} # ── 4. Backend meta ─────────────────────────────────────────────────────── from .state import _agent_tasks as _tasks _last_timing = { k: round(_TIMING_STORE[k][-1], 1) if _TIMING_STORE.get(k) else None for k in ("llm_first_token", "llm_total", "tool_call") } # ── overall status ──────────────────────────────────────────────────────── _all_ok = ai_section["ok"] and bool(sb_res.get("ok")) return { "ok": _all_ok, "status": "ok" if _all_ok else ("degraded" if ai_section["ok"] else "down"), "elapsed_ms": round((_t.monotonic() - t0) * 1000), "server_time": now, "version": "3.4.2", "ai": ai_section, "supabase": sb_res, "telegram": tg_res, "backend": { "active_tasks": len(_tasks), "heartbeat_status": hb.get("status"), "timing_last": _last_timing, "repair_stats": dict(_REPAIR_STATS), }, } # ── /api/auth/ping — verifica sync INTERNAL_TOKEN (pubblica, no auth richiesta) ── # Risponde immediatamente senza chiamate esterne. # Utile per verificare dall'iPhone se CF Worker e HF Space usano lo stesso token. @router.get('/api/auth/ping') async def auth_ping( x_internal_token: str | None = None, request: 'Request' = None, ): """ Endpoint pubblico per verificare il sync di INTERNAL_TOKEN. Logica: - internal_token_configured: True se INTERNAL_TOKEN è impostato come env var fissa (non generato al boot). Indica che HF Space ha il token configurato. - role_resolved: "MACHINE" se l'header X-Internal-Token in ingresso coincide con il token del server; "USER" altrimenti. - header_present: True se il chiamante ha inviato X-Internal-Token. Uso tipico da iPhone: curl https://arjanit98-terminal.hf.space/api/auth/ping → { "internal_token_configured": true, "role_resolved": "USER", ... } curl https://agente-ai.pages.dev/api/auth/ping → { "internal_token_configured": true, "role_resolved": "MACHINE", ... } (CF Worker aggiunge il token → role MACHINE se i due token coincidono) """ import os as _os server_token = _os.getenv('INTERNAL_TOKEN', '').strip() # Leggi header sia dal parametro sia dall'oggetto request (FastAPI può passare entrambi) hdr_token = x_internal_token if not hdr_token and request is not None: hdr_token = request.headers.get('X-Internal-Token') or request.headers.get('x-internal-token') hdr_token = (hdr_token or '').strip() configured = bool(server_token) role = 'MACHINE' if (configured and hdr_token and hdr_token == server_token) else 'USER' return { 'internal_token_configured': configured, 'role_resolved': role, 'header_present': bool(hdr_token), 'hint': ( 'Token OK — CF Worker e HF Space sono sincronizzati.' if role == 'MACHINE' else ( 'INTERNAL_TOKEN non impostato su HF Space — genera un token casuale ad ogni restart!' if not configured else 'Header X-Internal-Token assente o diverso dal token server — CF Worker non sincronizzato.' ) ), }