Spaces:
Running
Running
| """backend/api/providers.py β Health, tools, status, AI health, heartbeat (S354).""" | |
| import os, asyncio, time, logging | |
| import requests | |
| from fastapi import APIRouter, Request | |
| from fastapi import Depends | |
| from .auth_guard import require_role, AuthRole | |
| from .state import _sb, SENSITIVE, _ai_health_cache, _AI_HEALTH_TTL, _heartbeat_state, _TIMING_STORE, _REPAIR_STATS | |
| from .version import RUNTIME_VERSION | |
| 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 ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| async def health(): | |
| return { | |
| 'status': 'ok', | |
| 'version': RUNTIME_VERSION, | |
| 'supabase': _sb is not None, | |
| 'backend': 'HuggingFace Spaces / Railway', | |
| } | |
| # ββ S-DUAL-1: Load metrics for CF dual-space adaptive routing βββββββββββββββββ | |
| 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), | |
| } | |
| async def api_version(role: AuthRole = Depends(require_role(AuthRole.MACHINE))): # GAP-1-fix | |
| """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 | |
| }, | |
| } | |
| async def list_tools(role: AuthRole = Depends(require_role(AuthRole.MACHINE))): # GAP-1-fix | |
| 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)} | |
| # ββ P17-B2: Skill patterns β sync capabilities backend β frontend βββββββββββββ | |
| async def skills_patterns(request: Request, role: AuthRole = Depends(require_role(AuthRole.MACHINE))): # GAP-1-fix | |
| """ | |
| P17-B2: Capabilities del backend per sincronizzazione con subAgentRegistry.ts. | |
| Il frontend usa questo endpoint per sapere quali tool sono disponibili prima | |
| di assegnare subtask agli agenti, evitando routing verso tool non esistenti. | |
| Risposta: | |
| tools: lista tool con name, goal, description, required, risk, fallbacks | |
| tool_count: totale tool registrati | |
| version: versione backend | |
| ts: timestamp ms (cache busting) | |
| Sicurezza: richiede X-Internal-Token (coerente con /api/status e /api/version). | |
| """ | |
| _tok = os.getenv('INTERNAL_TOKEN', '') | |
| if _tok and request.headers.get('X-Internal-Token', '') != _tok: | |
| from fastapi.responses import JSONResponse as _JSONResp | |
| return _JSONResp({'error': 'Unauthorized β X-Internal-Token required'}, status_code=401) | |
| from tools.registry import TOOL_REGISTRY | |
| _tools = [] | |
| for _name, _info in TOOL_REGISTRY.items(): | |
| if _name.startswith("_"): | |
| continue | |
| _tools.append({ | |
| "name": _name, | |
| "goal": _info.get("goal", ""), | |
| "description": (_info.get("description") or "")[:300], | |
| "required": _info.get("required_inputs", []), | |
| "risk": _info.get("risk_level", "low"), | |
| "fallbacks": [ | |
| (f.get("name", "") if isinstance(f, dict) else str(f)) | |
| for f in _info.get("fallbacks", []) | |
| ], | |
| }) | |
| return { | |
| "tools": _tools, | |
| "tool_count": len(_tools), | |
| "version": "3.4.2", | |
| "ts": int(time.time() * 1000), | |
| } | |
| async def status(request: Request, role: AuthRole = Depends(require_role(AuthRole.MACHINE))): # GAP-1-fix | |
| # 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} | |
| async def health_manager_status(role: AuthRole = Depends(require_role(AuthRole.MACHINE))): | |
| """Ritorna lo stato del Health Manager (ARCH-P5.1).""" | |
| from .health_manager import health_manager | |
| return await health_manager.get_status() | |
| async def ai_provider_health(role: AuthRole = Depends(require_role(AuthRole.MACHINE))): # GAP-1-fix | |
| """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() | |
| def _classify_probe_error(exc: Exception) -> str: | |
| message = str(exc).lower() | |
| if "429" in message or "rate limit" in message or "quota" in message: | |
| return "rate_limit_or_quota" | |
| if "402" in message or "payment" in message or "credit" in message: | |
| return "credits_exhausted" | |
| if "401" in message or "403" in message or "unauthorized" in message or "forbidden" in message: | |
| return "authentication_or_permission" | |
| if "timeout" in message or "timed out" in message: | |
| return "timeout" | |
| if "404" in message or "not found" in message: | |
| return "model_or_endpoint_not_found" | |
| return "upstream_error" | |
| async def _openrouter_key_limits(provider) -> dict: | |
| if provider.name != "openrouter": | |
| return {} | |
| try: | |
| response = await asyncio.to_thread( | |
| requests.get, | |
| "https://openrouter.ai/api/v1/key", | |
| headers={"Authorization": f"Bearer {provider.api_key}"}, | |
| timeout=8, | |
| ) | |
| body = response.json() if response.content else {} | |
| data = body.get("data") if isinstance(body, dict) else {} | |
| if response.status_code >= 400: | |
| return {"key_status": response.status_code, "key_error_class": _classify_probe_error(RuntimeError(f"HTTP {response.status_code}"))} | |
| return { | |
| "key_status": response.status_code, | |
| "limit_remaining": data.get("limit_remaining"), | |
| "limit_reset": data.get("limit_reset"), | |
| "is_free_tier": data.get("is_free_tier"), | |
| "usage_daily": data.get("usage_daily"), | |
| } | |
| except Exception as exc: | |
| return {"key_error_class": _classify_probe_error(exc)} | |
| 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) | |
| result = {"name": provider.name, "profile": provider.profile, "ok": True, "status": "ok", "latency_ms": ms, | |
| "model": provider.default_model.split("/")[-1][:40]} | |
| result.update(await _openrouter_key_limits(provider)) | |
| return result | |
| except Exception as exc: | |
| ms = round((time.monotonic() - t0) * 1000) | |
| return {"name": provider.name, "profile": provider.profile, "ok": False, "status": "error", "latency_ms": ms, | |
| "error_class": _classify_probe_error(exc), | |
| "error": str(exc)[:300], "model": provider.default_model.split("/")[-1][:40], **(await _openrouter_key_limits(provider))} | |
| 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 ββββββββββββββββββββββββββββββββββ | |
| async def providers_canonical(role: AuthRole = Depends(require_role(AuthRole.MACHINE))): # GAP-1-fix | |
| """ | |
| 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 | |
| from .health_manager import health_manager | |
| 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) | |
| # ARCH-P5.1: Registra successo | |
| await health_manager.record_success(provider.name, ms) | |
| return {"name": provider.name, "ok": True, "latency_ms": ms} | |
| except Exception as exc: | |
| ms = round((time.monotonic() - t0) * 1000) | |
| # ARCH-P5.1: Registra fallimento | |
| await health_manager.record_failure(provider.name, str(exc), component_type="provider") | |
| return {"name": provider.name, "ok": False, "latency_ms": ms, "error": str(exc)[: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) | |
| await asyncio.sleep(_HEARTBEAT_INTERVAL_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()) | |
| # BUGFIX: eccezioni del heartbeat loop erano perse silenziosamente (crash invisibile) | |
| def _log_hb_exc(t): | |
| if not t.cancelled() and t.exception(): | |
| _logger.warning("[providers] heartbeat task raised: %s", t.exception()) | |
| _heartbeat_task.add_done_callback(_log_hb_exc) | |
| _logger.info("start_heartbeat: task avviato (pid=%s)", id(_heartbeat_task)) | |
| except Exception as exc: | |
| _logger.warning("start_heartbeat failed: %s", exc) | |
| async def debug_timing(role: AuthRole = Depends(require_role(AuthRole.MACHINE))): # GAP-1-fix | |
| """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"), | |
| } | |
| async def providers_heartbeat(role: AuthRole = Depends(require_role(AuthRole.MACHINE))): # GAP-1-fix | |
| now = int(time.time()) | |
| # Il primo ciclo async potrebbe non essere ancora partito: la route di health | |
| # deve restituire uno snapshot coerente, non propagare un KeyError come HTTP 500. | |
| return { | |
| "status": _heartbeat_state.get("status", "idle"), | |
| "best_provider": _heartbeat_state.get("best_provider"), | |
| "best_latency_ms": _heartbeat_state.get("best_latency_ms"), | |
| "providers": _heartbeat_state.get("providers", []), | |
| "last_run_at": _heartbeat_state.get("last_run_at"), | |
| "next_run_at": _heartbeat_state.get("next_run_at"), | |
| "runs": _heartbeat_state.get("runs", 0), | |
| "error": _heartbeat_state.get("error"), | |
| "interval_s": _HEARTBEAT_INTERVAL_S, | |
| "server_time": now, | |
| } | |
| # ββ /api/health/full β aggregated health (AI + Supabase + Telegram + backend) ββ | |
| 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 <hf-space-a-url>/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.' | |
| ) | |
| ), | |
| } | |
| # ββ P18-B2: /api/health/full β Health check dettagliato ββββββββββββββββββββββ | |
| async def health_full(role: AuthRole = Depends(require_role(AuthRole.MACHINE))): | |
| """ | |
| P18-B2: Health check dettagliato di tutti i sottosistemi. | |
| Auth: MACHINE (X-Internal-Token header obbligatorio). | |
| Esegue tutti i check in parallelo con timeout individuali (5s ciascuno). | |
| Risposta: | |
| status "ok" | "degraded" | "critical" | |
| checks { <nome>: { ok, latency_ms?, error?, ... } } | |
| HTTP 503 se almeno un check CRITICAL fallisce | |
| (CRITICAL = almeno un Supabase raggiungibile + env_config ok). | |
| Non-critical failures β HTTP 200 con status "degraded". | |
| """ | |
| from fastapi.responses import JSONResponse # P18-B2: local import β not in module scope | |
| import shutil as _shu | |
| _CHECK_TIMEOUT = 5.0 # secondi β timeout per ogni singolo check | |
| # ββ check: Supabase βββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| async def _ck_supabase(client: object, label: str) -> dict: | |
| if client is None: | |
| return {"ok": False, "error": "not configured"} | |
| t0 = time.monotonic() | |
| try: | |
| await asyncio.wait_for( | |
| asyncio.to_thread( | |
| lambda: client.table("agent_tasks").select("task_id").limit(1).execute() | |
| ), | |
| timeout=_CHECK_TIMEOUT, | |
| ) | |
| return {"ok": True, "latency_ms": round((time.monotonic() - t0) * 1000)} | |
| except asyncio.TimeoutError: | |
| return { | |
| "ok": False, "error": "timeout", | |
| "latency_ms": round((time.monotonic() - t0) * 1000), | |
| } | |
| except Exception as exc: | |
| return { | |
| "ok": False, "error": str(exc)[:200], | |
| "latency_ms": round((time.monotonic() - t0) * 1000), | |
| } | |
| # ββ check: Redis / Upstash ββββββββββββββββββββββββββββββββββββββββββββββββ | |
| async def _ck_redis() -> dict: | |
| import urllib.request as _ur, json as _js | |
| url = os.getenv("UPSTASH_REDIS_URL", "").strip() | |
| token = os.getenv("UPSTASH_REDIS_TOKEN", "").strip() | |
| if not url or not token: | |
| return {"ok": False, "error": "not configured"} | |
| t0 = time.monotonic() | |
| try: | |
| def _ping(): | |
| req = _ur.Request( | |
| f"{url}/ping", | |
| headers={"Authorization": f"Bearer {token}"}, | |
| method="GET", | |
| ) | |
| with _ur.urlopen(req, timeout=4) as r: | |
| return _js.loads(r.read()) | |
| resp = await asyncio.wait_for(asyncio.to_thread(_ping), timeout=_CHECK_TIMEOUT) | |
| ok = resp.get("result") == "PONG" | |
| return { | |
| "ok": ok, | |
| "latency_ms": round((time.monotonic() - t0) * 1000), | |
| **({} if ok else {"error": f"unexpected response: {resp}"}), | |
| } | |
| except asyncio.TimeoutError: | |
| return {"ok": False, "error": "timeout", | |
| "latency_ms": round((time.monotonic() - t0) * 1000)} | |
| except Exception as exc: | |
| return {"ok": False, "error": str(exc)[:200], | |
| "latency_ms": round((time.monotonic() - t0) * 1000)} | |
| # ββ check: LLM providers (da cache heartbeat β zero latenza) βββββββββββββ | |
| async def _ck_llm_providers() -> dict: | |
| cache = _ai_health_cache.get("data") | |
| if cache and cache.get("providers"): | |
| age_s = round(time.monotonic() - _ai_health_cache.get("at", 0)) | |
| available = [p for p in cache["providers"] if p.get("ok")] | |
| best = ( | |
| min(available, key=lambda p: p.get("latency_ms", 99_999)) | |
| if available else None | |
| ) | |
| return { | |
| "ok": len(available) > 0, | |
| "from_cache": True, | |
| "cache_age_s": age_s, | |
| "total": len(cache["providers"]), | |
| "available": len(available), | |
| "best_provider": best["name"] if best else None, | |
| "best_latency_ms": best["latency_ms"] if best else None, | |
| "providers": cache["providers"], | |
| } | |
| # Heartbeat non ancora eseguito (boot molto recente) | |
| return { | |
| "ok": True, | |
| "from_cache": False, | |
| "note": "heartbeat non ancora eseguito β dati disponibili dopo il primo ciclo (90s)", | |
| } | |
| # ββ check: Python exec ββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| async def _ck_exec_python() -> dict: | |
| import sys | |
| t0 = time.monotonic() | |
| try: | |
| proc = await asyncio.wait_for( | |
| asyncio.create_subprocess_exec( | |
| sys.executable, "-c", "print('ok')", | |
| stdout=asyncio.subprocess.PIPE, | |
| stderr=asyncio.subprocess.PIPE, | |
| ), | |
| timeout=_CHECK_TIMEOUT, | |
| ) | |
| stdout, _ = await asyncio.wait_for(proc.communicate(), timeout=_CHECK_TIMEOUT) | |
| ok = proc.returncode == 0 and stdout.strip() == b"ok" | |
| return { | |
| "ok": ok, | |
| "latency_ms": round((time.monotonic() - t0) * 1000), | |
| "python": sys.version.split()[0], | |
| **({} if ok else {"error": f"returncode={proc.returncode}"}), | |
| } | |
| except Exception as exc: | |
| return {"ok": False, "error": str(exc)[:200], | |
| "latency_ms": round((time.monotonic() - t0) * 1000)} | |
| # ββ check: JS sandbox (Deno β Node vm fallback) βββββββββββββββββββββββββββ | |
| async def _ck_js_sandbox() -> dict: | |
| t0 = time.monotonic() | |
| # Deno | |
| deno_candidates = [os.getenv("DENO_PATH", ""), "/root/.deno/bin/deno", "deno"] | |
| for c in deno_candidates: | |
| if not c: | |
| continue | |
| bin_path = c if (os.path.isabs(c) and os.path.isfile(c)) else _shu.which(c) | |
| if not bin_path: | |
| continue | |
| try: | |
| proc = await asyncio.wait_for( | |
| asyncio.create_subprocess_exec( | |
| bin_path, "eval", "console.log('ok')", | |
| stdout=asyncio.subprocess.PIPE, | |
| stderr=asyncio.subprocess.PIPE, | |
| ), | |
| timeout=_CHECK_TIMEOUT, | |
| ) | |
| stdout, _ = await asyncio.wait_for(proc.communicate(), timeout=_CHECK_TIMEOUT) | |
| if proc.returncode == 0 and b"ok" in stdout: | |
| return {"ok": True, "sandbox": "deno", "bin": bin_path, | |
| "latency_ms": round((time.monotonic() - t0) * 1000)} | |
| except Exception: | |
| pass | |
| # Node vm | |
| node_bin = _shu.which("node") | |
| if node_bin: | |
| try: | |
| proc = await asyncio.wait_for( | |
| asyncio.create_subprocess_exec( | |
| node_bin, "-e", | |
| "const vm=require('vm');vm.runInNewContext('1+1');console.log('ok')", | |
| stdout=asyncio.subprocess.PIPE, | |
| stderr=asyncio.subprocess.PIPE, | |
| ), | |
| timeout=_CHECK_TIMEOUT, | |
| ) | |
| stdout, _ = await asyncio.wait_for(proc.communicate(), timeout=_CHECK_TIMEOUT) | |
| if proc.returncode == 0 and b"ok" in stdout: | |
| return {"ok": True, "sandbox": "node_vm", "bin": node_bin, | |
| "latency_ms": round((time.monotonic() - t0) * 1000)} | |
| except Exception: | |
| pass | |
| return { | |
| "ok": False, | |
| "error": "nessun JS sandbox disponibile (Deno e Node non trovati)", | |
| "latency_ms": round((time.monotonic() - t0) * 1000), | |
| } | |
| # ββ check: Env config βββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| async def _ck_env_config() -> dict: | |
| critical = ["INTERNAL_TOKEN", "SUPABASE_URL", "SUPABASE_KEY"] | |
| important = ["UPSTASH_REDIS_URL", "ALLOWED_ORIGINS", "RESEND_API_KEY"] | |
| optional = ["OPERATOR_TOKEN", "ADMIN_TOKEN", | |
| "GROQ_API_KEY", "HF_TOKEN_A", "HF_TOKEN_B", "HF_TOKEN_C"] | |
| miss_crit = [v for v in critical if not os.getenv(v, "").strip()] | |
| miss_imp = [v for v in important if not os.getenv(v, "").strip()] | |
| miss_opt = [v for v in optional if not os.getenv(v, "").strip()] | |
| return { | |
| "ok": len(miss_crit) == 0, | |
| "missing_critical": miss_crit, | |
| "missing_important": miss_imp, | |
| "missing_optional": miss_opt, | |
| "railway_env": os.getenv("RAILWAY_ENVIRONMENT", ""), | |
| "space_role": os.getenv("SPACE_ROLE", ""), | |
| } | |
| # ββ check: Internal token sync ββββββββββββββββββββββββββββββββββββββββββββ | |
| async def _ck_internal_token() -> dict: | |
| tok = os.getenv("INTERNAL_TOKEN", "").strip() | |
| on_railway = bool( | |
| os.getenv("RAILWAY_ENVIRONMENT") or os.getenv("RAILWAY_PROJECT_ID") | |
| ) | |
| on_hf = bool(os.getenv("SPACE_ID") or os.getenv("HF_SPACE_ID")) | |
| configured = bool(tok) | |
| # Ephemeral = configurato solo in locale senza Railway/HF (generato al boot) | |
| ephemeral = configured and not (on_railway or on_hf) | |
| return { | |
| "ok": configured, | |
| "configured": configured, | |
| "ephemeral": ephemeral, | |
| "env": ("railway" if on_railway else ("hf_space" if on_hf else "local")), | |
| **({ | |
| "warning": "token generato al boot β CF Worker andrΓ aggiornato ad ogni riavvio" | |
| } if ephemeral else {}), | |
| } | |
| # ββ 3. Telegram β /getMe live (timeout 3s) ββββββββββββββββββββββββββββββββ | |
| async def _probe_telegram() -> dict: | |
| _pt = time.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((time.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((time.monotonic() - _pt) * 1000)} | |
| except Exception as exc: | |
| return {"ok": False, "configured": None, | |
| "error": str(exc)[:150], | |
| "latency_ms": round((time.monotonic() - _pt) * 1000)} | |
| # ββ check: Processo / Memoria βββββββββββββββββββββββββββββββββββββββββββββ | |
| async def _ck_process() -> dict: | |
| result: dict = {"ok": True, "uptime_s": round(time.monotonic() - _BOOT_TIME)} | |
| try: | |
| import psutil as _ps | |
| p = _ps.Process(os.getpid()) | |
| mem = p.memory_info() | |
| result.update({ | |
| "mem_rss_mb": round(mem.rss / 1024 / 1024, 1), | |
| "mem_vms_mb": round(mem.vms / 1024 / 1024, 1), | |
| "cpu_percent": p.cpu_percent(interval=0.05), | |
| "threads": p.num_threads(), | |
| }) | |
| except ImportError: | |
| result["mem_note"] = "psutil non disponibile" | |
| except Exception as exc: | |
| result["mem_error"] = str(exc)[:100] | |
| return result | |
| # ββ check: Disco ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| async def _ck_disk() -> dict: | |
| try: | |
| u = _shu.disk_usage("/") | |
| free_gb = round(u.free / 1024 ** 3, 2) | |
| total_gb = round(u.total / 1024 ** 3, 2) | |
| used_pct = round(u.used / u.total * 100, 1) | |
| ok = free_gb > 0.5 | |
| return { | |
| "ok": ok, | |
| "free_gb": free_gb, | |
| "total_gb": total_gb, | |
| "used_pct": used_pct, | |
| **({ | |
| "warning": f"spazio libero basso: {free_gb:.2f} GB" | |
| } if not ok else {}), | |
| } | |
| except Exception as exc: | |
| return {"ok": False, "error": str(exc)[:100]} | |
| # ββ Esegui tutti i check in parallelo βββββββββββββββββββββββββββββββββββββ | |
| from .state import _sb as _sb_h, _clients as _sb_clients_h | |
| # FIX-HEALTH-FULL: _sb2 e _sb_fallback non esistono in state.py. | |
| # Estraiamo i client dal pool _clients (A=primary, B=secondary, C=fallback). | |
| _sb2_h = _sb_clients_h[1]["client"] if len(_sb_clients_h) > 1 else None | |
| _sbf_h = _sb_clients_h[2]["client"] if len(_sb_clients_h) > 2 else None | |
| ( | |
| c_sb1, | |
| c_tg, | |
| c_sb2, | |
| c_sbf, | |
| c_redis, | |
| c_llm, | |
| c_py, | |
| c_js, | |
| c_env, | |
| c_tok, | |
| c_proc, | |
| c_disk, | |
| ) = await asyncio.gather( | |
| _ck_supabase(_sb_h, "primary"), | |
| _probe_telegram(), | |
| _ck_supabase(_sb2_h, "secondary"), | |
| _ck_supabase(_sbf_h, "fallback"), | |
| _ck_redis(), | |
| _ck_llm_providers(), | |
| _ck_exec_python(), | |
| _ck_js_sandbox(), | |
| _ck_env_config(), | |
| _ck_internal_token(), | |
| _ck_process(), | |
| _ck_disk(), | |
| ) | |
| checks = { | |
| "supabase_primary": c_sb1, | |
| "supabase_secondary": c_sb2, | |
| "supabase_fallback": c_sbf, | |
| "telegram": c_tg, | |
| "redis": c_redis, | |
| "llm_providers": c_llm, | |
| "exec_python": c_py, | |
| "exec_js_sandbox": c_js, | |
| "env_config": c_env, | |
| "internal_token": c_tok, | |
| "process": c_proc, | |
| "disk": c_disk, | |
| } | |
| # ββ Determina stato complessivo βββββββββββββββββββββββββββββββββββββββββββ | |
| # CRITICAL: almeno un Supabase deve rispondere + env_config deve essere ok | |
| supabase_any_ok = c_sb1["ok"] or c_sb2["ok"] or c_sbf["ok"] | |
| critical_ok = supabase_any_ok and c_env["ok"] | |
| # Non-critical: tutto il resto (GAP-UX-FIX: ignora redis/telegram non configurati) | |
| non_critical_failed = [ | |
| name for name, c in checks.items() | |
| if name not in ["env_config", "redis", "telegram"] and not c.get("ok") | |
| ] | |
| if not critical_ok: overall = "critical" | |
| elif non_critical_failed: overall = "degraded" | |
| else: overall = "ok" | |
| body = { | |
| "status": overall, | |
| "version": "3.4.2", | |
| "ts": int(time.time() * 1000), | |
| "checks": checks, | |
| "summary": { | |
| "supabase_any_ok": supabase_any_ok, | |
| "critical_ok": critical_ok, | |
| "degraded_checks": non_critical_failed, | |
| "total_checks": len(checks), | |
| "checks_ok": sum(1 for c in checks.values() if c.get("ok")), | |
| }, | |
| } | |
| if overall == "critical": | |
| return JSONResponse(status_code=503, content=body) | |
| return body | |
| # ββ S19-FIX: Endpoint per aggiornare modelli deprecati nella flotta βββββββββββ | |
| async def update_provider_models(role: AuthRole = Depends(require_role(AuthRole.MACHINE))): | |
| """S19: Aggiorna i modelli deprecati nella tabella ai_providers. | |
| Idempotente β sicuro da chiamare piΓΉ volte. | |
| Auth: MACHINE (X-Internal-Token obbligatorio). | |
| """ | |
| if _sb is None: | |
| return {"ok": False, "error": "Supabase non configurato", "updated": 0} | |
| # Mappa provider-specifica: (provider, modello_vecchio, modello_nuovo). | |
| # Lo stesso ID modello puΓ² essere valido su un provider e non su un altro: | |
| # filtrare per `name` evita di applicare un formato incompatibile alla riga sbagliata. | |
| # GPT-OSS 120B non compare come vecchio valore perchΓ© Γ¨ giΓ un modello supportato. | |
| MODEL_FIXES = [ | |
| ("groq", "llama-3.1-70b-versatile", "openai/gpt-oss-120b"), | |
| ("cerebras", "llama3.1-70b", "gpt-oss-120b"), | |
| ("nvidia", "llama-3.1-405b-instruct", "meta/llama-3.3-70b-instruct"), | |
| ("openrouter", "llama-3.1-405b", "openai/gpt-oss-20b:free"), | |
| ("sambanova", "llama3-70b", "DeepSeek-V3.2"), | |
| ("gemini", "gemini-1.5-flash", "gemini-3.5-flash-lite"), | |
| ("gemini", "gemini-1.5-pro", "gemini-3.6-flash"), | |
| ("openrouter", "claude-3.5-sonnet", "openai/gpt-oss-20b:free"), | |
| ] | |
| import asyncio as _aio | |
| total_updated = 0 | |
| results = [] | |
| for provider_name, old_model, new_model in MODEL_FIXES: | |
| try: | |
| r = await _aio.to_thread( | |
| lambda pn=provider_name, om=old_model, nm=new_model: _sb.table("ai_providers") | |
| .update({"default_model": nm}) | |
| .eq("name", pn) | |
| .eq("default_model", om) | |
| .execute() | |
| ) | |
| n = len(r.data) if r.data else 0 | |
| total_updated += n | |
| if n > 0: | |
| results.append({"provider": provider_name, "old": old_model, "new": new_model, "rows": n}) | |
| except Exception as exc: | |
| results.append({"provider": provider_name, "old": old_model, "new": new_model, "error": str(exc)[:100]}) | |
| # Disattiva provider E2B (non sono LLM provider) | |
| try: | |
| r_e2b = await _aio.to_thread( | |
| lambda: _sb.table("ai_providers") | |
| .update({"is_active": False}) | |
| .like("name", "e2b%") | |
| .eq("default_model", "base") | |
| .execute() | |
| ) | |
| n_e2b = len(r_e2b.data) if r_e2b.data else 0 | |
| if n_e2b > 0: | |
| results.append({"action": "deactivate_e2b", "rows": n_e2b}) | |
| except Exception as exc: | |
| results.append({"action": "deactivate_e2b", "error": str(exc)[:100]}) | |
| return { | |
| "ok": True, | |
| "total_updated": total_updated, | |
| "fixes": results, | |
| "message": f"Aggiornati {total_updated} provider con modelli deprecati.", | |
| } | |