models/ai_client.py

"""
ai_client.py — Real Provider Fleet (LLM API Ensemble)
Legge la flotta di provider LLM da Supabase (tabella `ai_providers`, se disponibile)
con fallback su variabili d'ambiente per gli 8 provider realmente configurati nel
progetto (Groq, OpenRouter, Cerebras, SambaNova, Gemini, NVIDIA NIM, OpenAI, HF Router).

STORIA: la versione precedente ("Dynamic SQL Fleet Edition") assumeva una flotta di
10 ZeroGPU Spaces (HF_SPACE_A1..E2) mai realmente esistita — nessuna di quelle env var
è mai stata configurata né documentata altrove nel repo, quindi self.providers era
sempre vuoto in produzione (ogni chiamata falliva silenziosamente con
"tutti i nodi A-E sono offline"). Sostituito con i provider realmente attivi
(vedi ENV_AUDIT_P21.md §6 e artifacts/agente-ai/src/config/env.ts).
"""
from __future__ import annotations

import asyncio
import json
import os
import re
import time as _time_mod
from dataclasses import dataclass
from typing import AsyncIterator, Optional, List, Tuple

from openai import OpenAI
from api.semantic_cache import get_cached_response, set_cached_response
import logging
_logger = logging.getLogger("agente_ai")

class ProviderUnavailableError(RuntimeError):
    """Raised when no configured LLM provider can produce a response.

    The error deliberately includes provider names only, never credentials or
    raw upstream payloads, so callers can distinguish infrastructure failure
    from a model answer without leaking sensitive data.
    """

    def __init__(self, providers: list[str] | tuple[str, ...]) -> None:
        self.providers = tuple(providers)
        detail = ", ".join(self.providers) if self.providers else "none"
        super().__init__(f"provider_unavailable: {detail}")


@dataclass(frozen=True)
class ProviderConfig:
    id: int = 0
    name: str = ""
    api_key: str = ""
    base_url: str = ""
    default_model: str = ""
    tier: int = 1
    purpose: str = "reasoning"
    profile: str = "general"

    @property
    def identity(self) -> tuple[str, str, str]:
        """Stable identity: different profiles must never share a client cache entry."""
        return (self.name, self.profile, self.base_url)

# Definizione statica dei provider LLM realmente attivi nel progetto.
# base_url punta sempre all'endpoint OpenAI-compatible ufficiale del provider
# (nessun proxy CF Worker qui: questo client gira lato backend Python, non browser).
_PROVIDER_DEFS = [
    # tier 0 — free tier veloce e affidabile
    {"name": "groq",       "env_key": "GROQ_API_KEY",       "base_url": "https://api.groq.com/openai/v1",                    "model_env": "GROQ_MODEL",      "default_model": "qwen/qwen3.6-27b",     "tier": 0, "purpose": "reasoning"},
    {"name": "cerebras",   "env_key": "CEREBRAS_API_KEY",   "base_url": "https://api.cerebras.ai/v1",                        "model_env": "CEREBRAS_MODEL",  "default_model": "gpt-oss-120b",                "tier": 0, "purpose": "reasoning"},
    {"name": "sambanova",  "env_key": "SAMBANOVA_API_KEY",  "base_url": "https://api.sambanova.ai/v1",                       "model_env": "SAMBANOVA_MODEL", "default_model": "DeepSeek-V3.1",               "tier": 0, "purpose": "reasoning"},
    # tier 1 — free tier con rate limit più stretti
    {"name": "openrouter", "env_key": "OPENROUTER_API_KEY", "base_url": "https://openrouter.ai/api/v1",                      "model_env": "OPENROUTER_MODEL","default_model": "openrouter/free", "tier": 1, "purpose": "coding"},
    {"name": "hf_router",  "env_key": "HF_TOKEN",           "base_url": "https://router.huggingface.co/v1",                  "model_env": "HF_MODEL",        "default_model": "Qwen/Qwen2.5-Coder-32B-Instruct", "tier": 1, "purpose": "coding"},
    {"name": "gemini",     "env_key": "GEMINI_API_KEY",     "base_url": "https://generativelanguage.googleapis.com/v1beta/openai/", "model_env": "GEMINI_MODEL", "default_model": "gemini-3.6-flash",        "tier": 1, "purpose": "memory"},
    # tier 2 — fallback opzionale (spesso a pagamento o quota limitata)
    {"name": "nvidia",     "env_key": "NVIDIA_API_KEY",     "base_url": "https://integrate.api.nvidia.com/v1",               "model_env": "NVIDIA_MODEL",    "default_model": "nvidia/nemotron-3-ultra-550b-a55b", "tier": 2, "purpose": "audit"},
]


class AIClient:
    def __init__(self, byok_credentials: dict[str, list[str]] | None = None, single_provider: bool = False) -> None:
        # Le chiavi BYOK appartengono a un singolo task e vivono solo in questa
        # istanza: non vengono scritte in env, Supabase, cache semantica o log.
        self._single_provider = single_provider
        self._byok_providers = self._providers_from_byok(byok_credentials or {})
        # I profili BYOK precedono i provider runtime: la stessa API conserva
        # comunque il fallback server-side in caso di quota o errore upstream.
        self.providers = self._byok_providers + self._load_providers()
        self._client_cache: dict[tuple[str, str, str], OpenAI] = {}
        # Round-robin e circuit breaker sono indicizzati per purpose e profilo.
        self._rr_indices: dict[str, int] = {}
        self._breaker: dict[tuple[str, str, str], dict[str, float | int]] = {}
        self._breaker_threshold = 2
        self._breaker_cooldown_s = 60.0

    @staticmethod
    def _providers_from_byok(credentials: dict[str, list[str]]) -> list[ProviderConfig]:
        """Build task-scoped provider profiles from browser BYOK credentials.

        Only known OpenAI-compatible providers are accepted. Values are copied
        into the transient client instance and intentionally never logged or
        persisted; empty, malformed and unsupported entries are ignored.
        """
        providers: list[ProviderConfig] = []
        for definition in _PROVIDER_DEFS:
            raw_tokens = credentials.get(definition["name"], [])
            if not isinstance(raw_tokens, list):
                continue
            seen_tokens: set[str] = set()
            for raw_token in raw_tokens:
                if not isinstance(raw_token, str):
                    continue
                token = raw_token.strip()
                if not token or token in seen_tokens:
                    continue
                seen_tokens.add(token)
                providers.append(ProviderConfig(
                    id=-(10_000 + len(providers)),
                    name=definition["name"],
                    api_key=token,
                    base_url=definition["base_url"],
                    default_model=os.getenv(definition["model_env"], definition["default_model"]),
                    tier=definition["tier"],
                    purpose=definition["purpose"],
                    profile=f"byok-{len(seen_tokens)}",
                ))
        return providers

    def _load_providers(self) -> list[ProviderConfig]:
        """Carica la flotta: prova Supabase (tabella `ai_providers`, source of
        truth dichiarata in supabase/migrations/20260711_ai_providers_fleet.sql),
        fallback sui provider reali via env se Supabase non è raggiungibile/vuoto
        (es. progetto sospeso per fatturazione, tabella non ancora popolata)."""
        database_providers = self._try_load_from_supabase()
        environment_profiles = [
            profile
            for definition in _PROVIDER_DEFS
            for profile in self._profile_rows_from_env(definition)
        ]
        if environment_profiles:
            profiled_names = {profile.name for profile in environment_profiles}
            # Explicit profile pools override a same-provider Supabase credential;
            # DB providers not covered by a pool remain available as fallbacks.
            database_providers = [
                provider for provider in database_providers
                if provider.name not in profiled_names
            ]
            return environment_profiles + database_providers
        return database_providers or self._discover_providers_from_env()

    @staticmethod
    def _runtime_model_override(row: dict) -> str:
        """Use an explicit model environment override for a known provider endpoint.

        Supabase remains the source for provider credentials and ordering; runtime
        model-selection variables deliberately win so emergency model migrations
        do not require reading or mutating provider secrets in the database.
        """
        database_model = str(row.get("default_model", "")).strip()
        row_base_url = str(row.get("base_url", "")).rstrip("/")
        for definition in _PROVIDER_DEFS:
            if row_base_url == definition["base_url"].rstrip("/"):
                # Explicit runtime configuration always wins over persisted DB values.
                configured_model = os.getenv(definition["model_env"], "").strip()
                if configured_model:
                    return configured_model
                # Supabase may retain a model retired by the provider. Do not let a
                # stale row override the tested repository default after deployment.
                if (
                    definition["name"] == "groq"
                    and database_model in {
                        "llama-3.3-70b-versatile",
                        "llama-3.1-70b-versatile",
                        "llama-3.1-8b-instant",
                    }
                ):
                    return definition["default_model"]
                return database_model or definition["default_model"]
        return database_model

    @staticmethod
    def _is_legacy_schema_error(exc: Exception) -> bool:
        """Riconosce il layout `ai_providers` precedente alla flotta canonica.

        Quel layout espone `model_name`, `priority` e `provider_type`, ma
        contiene record storici e modelli deprecati. Fino alla migrazione non va
        promosso a source of truth: il fallback ambiente aggiornato è più sicuro.
        """
        message = str(exc).lower()
        return (
            "column ai_providers." in message
            and "does not exist" in message
            and any(column in message for column in (
                "default_model", "tier", "purpose", "success_count",
            ))
        )

    @staticmethod
    def _profile_rows_from_env(definition: dict) -> list[ProviderConfig]:
        """Load optional per-provider profiles without logging secret values.

        Format: ``<PROVIDER>_PROFILES_JSON=[{"profile":"p1","api_key":"...", "model":"..."}]``.
        The legacy single-key variable remains supported and is loaded after profiles.
        """
        env_name = f"{definition['name'].upper()}_PROFILES_JSON"
        raw = os.getenv(env_name, "").strip()
        if not raw:
            return []
        try:
            rows = json.loads(raw)
        except json.JSONDecodeError:
            _logger.warning("AIClient: %s non valido, profili ignorati", env_name)
            return []
        if not isinstance(rows, list):
            _logger.warning("AIClient: %s deve essere un array JSON", env_name)
            return []
        result: list[ProviderConfig] = []
        for index, row in enumerate(rows):
            if not isinstance(row, dict) or not row.get("api_key"):
                continue
            result.append(ProviderConfig(
                id=-(index + 1),
                name=definition["name"],
                api_key=str(row["api_key"]),
                base_url=str(row.get("base_url") or definition["base_url"]),
                default_model=str(row.get("model") or os.getenv(definition["model_env"], definition["default_model"])),
                tier=definition["tier"],
                purpose=str(row.get("purpose") or definition["purpose"]),
                profile=str(row.get("profile") or f"profile-{index + 1}"),
            ))
        return result

    def _try_load_from_supabase(self) -> list[ProviderConfig]:
        url = os.getenv("SUPABASE_URL", "")
        key = os.getenv("SUPABASE_SERVICE_ROLE_KEY", "")
        if not url or not key:
            return []
        try:
            from supabase import create_client
            sb = create_client(url, key)
            try:
                res = (
                    sb.table("ai_providers")
                    .select("id,name,api_key,base_url,default_model,tier,purpose")
                    .eq("is_active", True)
                    .order("tier", desc=False)
                    .order("success_count", desc=True)
                    .execute()
                )
            except Exception as exc:
                if self._is_legacy_schema_error(exc):
                    # Non usare il layout storico: contiene provider fittizi e
                    # modelli superati. La migrazione normalizzerà la tabella;
                    # nel frattempo il caller seleziona il fallback env corrente.
                    return []
                raise
            rows = res.data or []
            return [
                ProviderConfig(
                    id=row["id"], name=row["name"], api_key=row["api_key"],
                    base_url=row["base_url"], default_model=self._runtime_model_override(row),
                    tier=row["tier"], purpose=row["purpose"],
                    # Legacy schema has no profile column: the row id is still a
                    # stable profile identity and prevents client-cache collisions.
                    profile=f"db-{row['id']}",
                )
                for row in rows
            ]
        except Exception as e:
            _logger.warning(f"AIClient: Supabase ai_providers non disponibile, uso fallback env: {e}")
            return []

    def _discover_providers_from_env(self) -> list[ProviderConfig]:
        """Discovery dei provider LLM realmente configurati via env var (vedi
        ENV_AUDIT_P21.md §6). Un provider viene incluso solo se la sua API key
        è impostata — nessun placeholder, nessun nodo fantasma."""
        providers = []
        for i, d in enumerate(_PROVIDER_DEFS):
            providers.extend(self._profile_rows_from_env(d))
            api_key = os.getenv(d["env_key"], "")
            if not api_key:
                continue
            providers.append(ProviderConfig(
                id=i,
                name=d["name"],
                api_key=api_key,
                base_url=d["base_url"],
                default_model=os.getenv(d["model_env"], d["default_model"]),
                tier=d["tier"],
                purpose=d["purpose"],
                profile="legacy",
            ))
        if not providers:
            _logger.error("AIClient: nessuna API key provider configurata (Groq/OpenRouter/Cerebras/SambaNova/Gemini/NVIDIA/HF_TOKEN tutte assenti)")
        return providers

    def _client_for(self, provider: ProviderConfig) -> OpenAI:
        if provider.identity not in self._client_cache:
            self._client_cache[provider.identity] = OpenAI(
                api_key=provider.api_key,
                base_url=provider.base_url,
                # I task coding possono richiedere più di 20 s prima del primo
                # chunk dal fallback gratuito; il budget esterno resta finito.
                timeout=45,
                max_retries=0
            )
        return self._client_cache[provider.identity]

    def _is_available(self, provider: ProviderConfig) -> bool:
        state = self._breaker.get(provider.identity)
        return not state or float(state.get("open_until", 0.0)) <= _time_mod.monotonic()

    def _record_success(self, provider: ProviderConfig) -> None:
        self._breaker.pop(provider.identity, None)

    def _record_failure(self, provider: ProviderConfig, exc: Exception) -> None:
        message = str(exc).lower()
        if not any(token in message for token in ("401", "403", "429", "500", "502", "503", "504", "rate limit", "quota")):
            return
        state = self._breaker.setdefault(provider.identity, {"failures": 0, "open_until": 0.0})
        failures = int(state.get("failures", 0)) + 1
        severe = any(token in message for token in ("401", "403"))
        quota_limited = any(token in message for token in ("429", "rate limit", "quota"))
        # A quota/rate-limit response is deterministic: retrying the same
        # profile immediately only creates a storm. Open that profile on the
        # first signal and let the provider pool move to another provider.
        threshold = 1 if severe or quota_limited else self._breaker_threshold
        if failures >= threshold:
            cooldown = 900.0 if severe else self._rate_limit_cooldown_seconds(message) if quota_limited else self._breaker_cooldown_s
            state["open_until"] = _time_mod.monotonic() + cooldown
        state["failures"] = failures

    @staticmethod
    def _rate_limit_cooldown_seconds(message: str) -> float:
        """Return a provider reset-aware cooldown, never shorter than 15 min."""
        reset_match = re.search(r"x-ratelimit-reset[^0-9]*(\d{10,13})", message, re.IGNORECASE)
        if reset_match:
            reset_value = float(reset_match.group(1))
            reset_epoch = reset_value / 1000.0 if reset_value > 10_000_000_000 else reset_value
            return max(900.0, reset_epoch - _time_mod.time())
        return 900.0

    def _execution_pool(self, providers: list[ProviderConfig], purpose: str) -> list[ProviderConfig]:
        """Return one rotated, healthy profile per provider endpoint group."""
        groups: dict[tuple[str, str], list[ProviderConfig]] = {}
        for provider in providers:
            if not self._is_available(provider):
                continue
            groups.setdefault((provider.name, provider.base_url), []).append(provider)
        selected: list[ProviderConfig] = []
        for group_key, profiles in groups.items():
            index_key = f"{purpose}:{group_key[0]}:{group_key[1]}"
            start = self._rr_indices.get(index_key, 0)
            selected.append(profiles[start % len(profiles)])
            self._rr_indices[index_key] = start + 1
        return selected

    def _inter_provider_fallback_pool(
        self,
        purpose: str,
        excluded: set[str] | None = None,
        providers: list[ProviderConfig] | None = None,
    ) -> list[ProviderConfig]:
        """Select one healthy profile per provider, prioritizing the target purpose.

        A provider whose complete profile group is open in the circuit breaker is
        absent from this list; the next healthy provider becomes the automatic
        fallback. This prevents retry storms against an exhausted pool.
        """
        excluded = excluded or set()
        source = self.providers if providers is None else providers
        candidates = [
            provider for provider in source
            if provider.name not in excluded and self._is_available(provider)
        ]
        candidates.sort(key=lambda provider: (
            0 if provider.purpose == purpose else 1,
            provider.tier,
            provider.name,
            provider.profile,
        ))
        return self._execution_pool(candidates, f"fallback:{purpose}")

    async def _fetch_one(self, provider: ProviderConfig, messages: list, temperature: float, max_tokens: int) -> Tuple[ProviderConfig, str, float]:
        start = _time_mod.monotonic()
        try:
            client = self._client_for(provider)
            response = await asyncio.wait_for(
                asyncio.to_thread(
                    client.chat.completions.create, 
                    model=provider.default_model, 
                    messages=messages, 
                    temperature=temperature, 
                    max_tokens=max_tokens,
                    **({"reasoning_effort": "none"} if provider.name == "groq" and provider.default_model == "qwen/qwen3.6-27b" else {})
                ),
                # Il fallback non-streaming deve avere lo stesso budget del client:
                # 15s scartava provider sani su richieste coding che richiedono
                # più tempo per produrre una risposta completa dopo uno stream interrotto.
                timeout=45
            )
            self._record_success(provider)
            return provider, response.choices[0].message.content or "", _time_mod.monotonic() - start
        except Exception as e:
            self._record_failure(provider, e)
            _logger.warning(f"Provider {provider.name}/{provider.profile} fallito: {e}")
            return provider, f"ERROR: {str(e)}", 0.0

    def _get_round_robin_provider(self, purpose: str) -> Optional[ProviderConfig]:
        """Ritorna uno dei provider con lo stesso purpose usando round-robin."""
        purpose_providers = [p for p in self.providers if p.purpose == purpose]
        if not purpose_providers:
            return None
        
        idx = self._rr_indices.get(purpose, 0)
        provider = purpose_providers[idx % len(purpose_providers)]
        self._rr_indices[purpose] = idx + 1
        return provider

    async def chat(self, messages: list, model: Optional[str] = None, temperature: float = 0.7, max_tokens: int = 4096) -> str:
        if not self.providers: self.providers = self._byok_providers + self._load_providers()
        
        # S-CACHE-1: Lookup semantico preventivo
        cached = await get_cached_response(messages)
        if cached:
            _logger.info("[fleet] Semantic Cache HIT — risparmio quota provider")
            return cached

        # 1. Determina l'intento per il routing iniziale (specializzazione per purpose,
        #    non per nodo hardware: i provider reali non sono dedicati per profilo).
        prompt = str(messages[-1].get("content", "")).lower()
        if any(k in prompt for k in ["code", "python", "fix", "script"]):
            primary_purpose = "coding"
        elif any(k in prompt for k in ["cerca", "ricorda", "database", "memory"]):
            primary_purpose = "memory"
        elif any(k in prompt for k in ["sicuro", "audit", "verifica", "check"]):
            primary_purpose = "audit"
        else:
            primary_purpose = "reasoning"

        # 2. Costruisce il pool di esecuzione (provider del purpose target + fallback globale)
        pool = [p for p in self.providers if p.purpose == primary_purpose]
        
        # Se il pool è vuoto, prendiamo i primi 4 provider disponibili (per tier)
        if not pool:
            pool = self.providers[:4]
        
        if not pool:
            raise ProviderUnavailableError([])
        
        # 3. Un solo profilo per endpoint e richiesta: round-robin evita che
        # profili condividano quota e client, mentre provider diversi restano
        # disponibili come ensemble/fallback.
        pool = self._execution_pool(pool, primary_purpose)
        # Public API requests use one provider at a time. The previous ensemble
        # fan-out multiplied outbound connections per request and exhausted the
        # small Space under concurrent load. Fallback remains available below.
        if self._single_provider:
            pool = pool[:1]
        results = []
        if pool:
            tasks = [self._fetch_one(p, messages, temperature, max_tokens) for p in pool]
            results = await asyncio.gather(*tasks)

        valid = [result for result in results if not result[1].startswith("ERROR:") and len(result[1]) > 10]
        if valid:
            best_r = self._judge_best_response(results, primary_purpose)
        else:
            # Il pool primario è interamente in rate limit, errore auth o timeout:
            # prova un solo profilo per ogni provider sano, in ordine di purpose/tier.
            excluded = {provider.name for provider, _response, _latency in results}
            fallback_pool = self._inter_provider_fallback_pool(primary_purpose, excluded)
            fallback_results = []
            for fallback in fallback_pool:
                result = await self._fetch_one(fallback, messages, temperature, max_tokens)
                fallback_results.append(result)
                if not result[1].startswith("ERROR:") and len(result[1]) > 10:
                    _logger.info(
                        "[fleet] inter-provider fallback succeeded on %s/%s",
                        fallback.name,
                        fallback.profile,
                    )
                    best_r = result[1]
                    break
            else:
                failed_names = [provider.name for provider, _response, _latency in results + fallback_results]
                raise ProviderUnavailableError(failed_names)

        # S-CACHE-1: Popolamento cache asincrono
        if not best_r.startswith("🔴"):
            asyncio.create_task(set_cached_response(messages, best_r))

        return best_r

    def _judge_best_response(self, results: List[Tuple[ProviderConfig, str, float]], target_purpose: str) -> str:
        valid = [(p, r, t) for p, r, t in results if not r.startswith("ERROR:") and len(r) > 10]
        if not valid:
            raise ProviderUnavailableError([p.name for p, _r, _t in results])
        
        def score(item):
            p, r, t = item
            s = len(r)
            # Bonus purpose target (Specializzazione)
            if p.purpose == target_purpose: s += 2000
            # Bonus latenza (Performance mobile-first)
            if t < 2.0: s += 500
            if "```" in r: s += 300
            return s

        _, best_r, _ = max(valid, key=score)
        return best_r

    async def stream_chat(self, messages: list, model: Optional[str] = None, temperature: float = 0.7, max_tokens: int = 4096) -> AsyncIterator[str]:
        """Streaming con failover immediato tra i provider configurati."""
        if not self.providers: self.providers = self._byok_providers + self._load_providers()
        # Nessun provider configurato: feedback immediato all'utente invece di
        # cadere silenziosamente nel loop vuoto e dare un messaggio generico.
        if not self.providers:
            raise ProviderUnavailableError([])
        
        # Un client di ruolo può contenere un solo provider specializzato.
        # Dopo il suo primario, integra la flotta runtime non duplicata: un limite
        # temporaneo di quel provider non deve rendere indisponibile l'intero task.
        providers = list(self.providers)
        try:
            for fallback in self._load_providers():
                if not any(
                    current.name == fallback.name
                    and current.base_url == fallback.base_url
                    for current in providers
                ):
                    providers.append(fallback)
        except Exception as exc:
            _logger.debug("Streaming fleet expansion skipped: %s", type(exc).__name__)

        # Un profilo sano per provider: se l’intero pool primario è in rate
        # limit, il fallback passa automaticamente al provider successivo.
        providers = self._inter_provider_fallback_pool("stream", providers=providers)
        attempted: list[str] = []
        for provider in providers:
            attempted.append(provider.name)
            emitted = False
            try:
                client = self._client_for(provider)
                stream = await asyncio.to_thread(
                    client.chat.completions.create,
                    model=provider.default_model,
                    messages=messages,
                    temperature=temperature,
                    max_tokens=max_tokens,
                    stream=True,
                    **({"reasoning_effort": "none"} if provider.name == "groq" and provider.default_model == "qwen/qwen3.6-27b" else {}),
                )
                iterator = iter(stream)
                while True:
                    chunk = await asyncio.to_thread(next, iterator, None)
                    if chunk is None:
                        break
                    if chunk.choices and chunk.choices[0].delta.content:
                        emitted = True
                        yield chunk.choices[0].delta.content
                self._record_success(provider)
                return
            except Exception as e:
                self._record_failure(provider, e)
                _logger.warning(
                    "Streaming fallito su %s/%s (emitted=%s): %s",
                    provider.name, provider.profile, emitted, e,
                )
                # Retry solo prima del primo chunk: dopo output parziale un
                # retry produrrebbe testo duplicato o una risposta incoerente.
                if emitted:
                    raise
                continue

        raise ProviderUnavailableError(attempted)