api/execution_fabric.py

"""
backend/api/execution_fabric.py — Execution Fabric (ARCH-K2.5 + ARCH-K2.6 Health Manager)

CRIT-B fix: circuit breaker Oracle (3 errori → blocco 300s) + fallback Railway con auth completa.
CRIT-C fix: parser 402/429 robusto — body scan + pattern estesi nell'exception handler.
"""
from __future__ import annotations
import asyncio
import logging
import os
import time
import uuid
from enum import Enum
from typing import Any, Optional
import httpx
from fastapi import APIRouter, Depends
from pydantic import BaseModel, Field

from .auth_guard import AuthRole, require_role
from .token_rotator import rotator as _token_rotator

try:
    from .telemetry import record_kernel_event as _rke
except Exception:
    def _rke(*_a, **_kw): pass

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

# ── Configurazione Timing (ARCH-T1.1) ──────────────────────────────────────────
_DEFAULT_TIMEOUT_S = 30.0     # Timeout standard per chiamate API
_ORACLE_TIMEOUT_S  = 60.0     # Oracle ha più tempo per il calcolo pesante
_HF_WARMUP_S      = 10.0     # Tempo di attesa se lo Space è in sleep
_RETRY_DELAY_S     = 1.5      # Attesa tra i tentativi di rotazione token
_HEALTH_CHECK_INT  = 120.0    # Intervallo health check in background

# ── CRIT-B: Circuit Breaker Oracle ─────────────────────────────────────────────
_ORACLE_CB_THRESHOLD  = 3      # errori consecutivi prima di aprire il circuit
_ORACLE_CB_TIMEOUT_S  = 300.0  # secondi di blocco dopo apertura (5 min)

# ── CRIT-C: pattern quota/rate-limit (body + eccezioni) ───────────────────────
_QUOTA_PATTERNS = (
    "quota", "egress", "rate limit", "rate_limit", "billing",
    "402", "429", "limit exceeded", "credits", "insufficient",
    "payment", "upgrade", "hours", "quota_or_ratelimit",
)


class ProviderKind(str, Enum):
    HF_SPACE   = "hf_space"
    # RAILWAY rimosso
    DOCKER     = "docker"
    ORACLE     = "oracle"
    LOCAL      = "local"

class ProviderHealth(str, Enum):
    OK        = "ok"
    DEGRADED  = "degraded"
    DOWN      = "down"
    UNKNOWN   = "unknown"

class AlwaysOn(str, Enum):
    YES       = "yes"
    ON_DEMAND = "on-demand"
    NO        = "no"

class ProviderSpec(BaseModel):
    provider_id:    str
    name:           str
    kind:           ProviderKind   = ProviderKind.LOCAL
    base_url:       str            = ""
    capabilities:   list[str]      = Field(default_factory=list)
    gpu:            bool           = False
    always_on:      AlwaysOn       = AlwaysOn.ON_DEMAND
    max_concurrency: int           = 10
    priority:       int            = 50
    cost_unit:      float          = 0.0
    region:         str            = "eu"
    timeout:        float          = _DEFAULT_TIMEOUT_S

class ProviderState(BaseModel):
    provider_id:        str
    health:             ProviderHealth      = ProviderHealth.UNKNOWN
    active_tasks:       int                 = 0
    last_check_ts:      float               = 0.0
    error_count:        int                 = 0

class DispatchRequest(BaseModel):
    capability:     str
    task_id:        str   = Field(default_factory=lambda: str(uuid.uuid4()))
    payload:        dict  = Field(default_factory=dict)
    require_gpu:    bool  = False
    require_isolation: bool = False
    prefer_region:  str | None = None
    provider_hint:  str | None = None
    correlation_id: str   = Field(default_factory=lambda: str(uuid.uuid4()))

class DispatchResult(BaseModel):
    task_id:        str
    provider_id:    str
    provider_name:  str
    provider_kind:  str
    capability:     str
    status:         str
    response:       dict        = Field(default_factory=dict)
    latency_ms:     float       = 0.0
    correlation_id: str         = ""


class ExecutionFabric:
    def __init__(self):
        self._specs:  dict[str, ProviderSpec]  = {}
        self._states: dict[str, ProviderState] = {}
        self._initialized = False

        # CRIT-B: circuit breaker state per provider (in-memory, per istanza)
        # { provider_id: {"errors": int, "blocked_until": float} }
        self._circuit: dict[str, dict] = {}

    # ── Inizializzazione provider ─────────────────────────────────────────────

    async def initialize(self):
        if self._initialized:
            return
        self._initialized = True

        providers: list[ProviderSpec] = []

        # Local fallback — sempre disponibile, nessun HTTP
        providers.append(ProviderSpec(
            provider_id="local",
            name="Local",
            kind=ProviderKind.LOCAL,
            base_url="",
            capabilities=["exec", "tool", "shell"],
            priority=10,
            always_on=AlwaysOn.YES,
        ))

        # HF Space principale — no-op se HF_SPACE_URL non configurata
        hf_url = os.getenv("HF_SPACE_URL", "").rstrip("/")
        if hf_url:
            providers.append(ProviderSpec(
                provider_id="hf-space-main",
                name="HF Space Main",
                kind=ProviderKind.HF_SPACE,
                base_url=hf_url,
                capabilities=["exec", "tool", "llm", "browse"],
                priority=50,
                always_on=AlwaysOn.ON_DEMAND,
                timeout=_DEFAULT_TIMEOUT_S,
            ))

        # Railway core backend rimosso (migrato su HF Spaces)

        # Oracle Cloud VM — no-op se ORACLE_CLOUD_VM_URL non impostata
        oracle_url = os.getenv("ORACLE_CLOUD_VM_URL", "").rstrip("/")
        if oracle_url:
            providers.append(ProviderSpec(
                provider_id="oracle-cloud-vm-01",
                name="Oracle Cloud VM",
                kind=ProviderKind.ORACLE,
                base_url=oracle_url,
                capabilities=["exec", "tool", "llm", "gpu"],
                gpu=True,
                priority=80,
                always_on=AlwaysOn.ON_DEMAND,
                timeout=_ORACLE_TIMEOUT_S,
            ))

        for spec in providers:
            self._specs[spec.provider_id]  = spec
            self._states[spec.provider_id] = ProviderState(provider_id=spec.provider_id)
            self._circuit[spec.provider_id] = {"errors": 0, "blocked_until": 0.0}

        _logger.info(
            "[fabric] Providers registrati: %s",
            ", ".join(f"{p.provider_id}({p.kind.value})" for p in providers)
        )

    # ── CRIT-B: Circuit Breaker helpers ──────────────────────────────────────

    def _is_circuit_open(self, provider_id: str) -> bool:
        """True se il provider è in blocco circuit breaker."""
        cb = self._circuit.get(provider_id, {})
        blocked_until = cb.get("blocked_until", 0.0)
        if blocked_until > time.time():
            return True
        # Reset automatico dopo il timeout
        if blocked_until > 0.0:
            self._circuit[provider_id]["errors"] = 0
            self._circuit[provider_id]["blocked_until"] = 0.0
            _logger.info("[fabric] Circuit breaker RESET per %s", provider_id)
        return False

    def _record_oracle_error(self, provider_id: str) -> bool:
        """
        Registra un errore Oracle. Se si supera la soglia, apre il circuit.
        Ritorna True se il circuit è stato appena aperto.
        """
        cb = self._circuit.setdefault(provider_id, {"errors": 0, "blocked_until": 0.0})
        cb["errors"] += 1
        if cb["errors"] >= _ORACLE_CB_THRESHOLD:
            cb["blocked_until"] = time.time() + _ORACLE_CB_TIMEOUT_S
            _logger.error(
                "[fabric] Circuit breaker APERTO per %s (%d errori consecutivi) "
                "— blocco per %.0fs fino a %s",
                provider_id, cb["errors"], _ORACLE_CB_TIMEOUT_S,
                time.strftime("%H:%M:%S", time.localtime(cb["blocked_until"]))
            )
            return True
        return False

    def _reset_oracle_errors(self, provider_id: str) -> None:
        """Azzera il contatore errori dopo un successo."""
        if provider_id in self._circuit:
            self._circuit[provider_id]["errors"] = 0

    # ── Selezione provider ────────────────────────────────────────────────────

    def _cb_allow(self, provider_id: str) -> bool:
        return True

    def _select(self, capability: str, exclude: set[str] | None = None) -> Optional[str]:
        """
        Seleziona il provider con la priorità più alta che:
        - supporta la capability richiesta
        - non è DOWN
        - non è in circuit breaker aperto
        - non è nella lista exclude
        Ordine: priority DESC (higher = preferred).
        """
        exclude = exclude or set()
        candidates = [
            (spec.priority, pid, spec)
            for pid, spec in self._specs.items()
            if capability in spec.capabilities
            and self._states[pid].health != ProviderHealth.DOWN
            and not self._is_circuit_open(pid)
            and pid not in exclude
        ]
        if not candidates:
            return None
        # Ordina per priority decrescente
        candidates.sort(key=lambda x: x[0], reverse=True)
        return candidates[0][1]

    # ── CRIT-C: rilevamento quota/rate-limit ──────────────────────────────────

    @staticmethod
    def _is_quota_error(text: str) -> bool:
        """True se il testo (body o eccezione) indica quota/rate-limit."""
        t = text.lower()
        return any(p in t for p in _QUOTA_PATTERNS)

    # ── Chiamata HTTP al provider ─────────────────────────────────────────────

    async def _call_provider(self, spec: ProviderSpec, req: DispatchRequest) -> dict:
        timeout = _ORACLE_TIMEOUT_S if spec.kind == ProviderKind.ORACLE else spec.timeout

        # Provider locale: nessuna chiamata HTTP
        if spec.kind == ProviderKind.LOCAL:
            return {"status": "ok", "provider": "local", "task_id": req.task_id}

        if not spec.base_url:
            raise ValueError(f"base_url non configurato per provider {spec.provider_id}")

        # ── Header auth per provider kind ────────────────────────────────────
        headers: dict[str, str] = {"Content-Type": "application/json"}
        internal_token  = os.getenv("INTERNAL_TOKEN", "")
        hf_token        = os.getenv("HF_TOKEN", "")
        # Railway auth rimosso
        if spec.kind in (ProviderKind.ORACLE, ProviderKind.DOCKER):
            if internal_token:
                headers["X-Internal-Token"] = internal_token

        payload = {
            "task_id":        req.task_id,
            "capability":     req.capability,
            "payload":        req.payload,
            "correlation_id": req.correlation_id,
        }
        endpoint = f"{spec.base_url.rstrip('/')}/api/exec"

        async with httpx.AsyncClient(timeout=timeout) as client:
            resp = await client.post(endpoint, json=payload, headers=headers)

            # CRIT-C: rileva quota/rate-limit sia da status code che da body
            body_text = resp.text
            if resp.status_code in (402, 429) or self._is_quota_error(body_text):
                _logger.warning(
                    "[fabric] %s — quota/rate-limit rilevato (HTTP %s)",
                    spec.provider_id, resp.status_code
                )
                return {
                    "status_code": resp.status_code,
                    "error":       "quota_or_ratelimit",
                    "body":        body_text[:200],
                }

            resp.raise_for_status()
            try:
                return resp.json()
            except Exception:
                return {"status": "ok", "raw": body_text[:500]}

    # ── Health management ────────────────────────────────────────────────────

    def _update_health(self, pid: str, health: ProviderHealth) -> None:
        if pid in self._states:
            self._states[pid].health = health

    # ── Dispatch pubblico (con chunking) ────────────────────────────────────

    async def dispatch(self, req: DispatchRequest) -> DispatchResult:
        await self.initialize()

        # --- LOGICA CHUNKING (S-CHUNK) ---
        from tools.payload_chunker import chunk_payload
        chunks = chunk_payload(req.payload, max_kb=450)

        if len(chunks) > 1:
            _logger.info("[fabric] Payload grande — suddivisione in %d pezzi.", len(chunks))
            final_responses = []
            for chunk in chunks:
                chunk_req = req.model_copy(update={"payload": chunk})
                res = await self._dispatch_single(chunk_req)
                final_responses.append(res.response)
            return DispatchResult(
                task_id=req.task_id, provider_id="multi", provider_name="Fabric Chunker",
                provider_kind="internal", capability=req.capability, status="executed",
                response={"chunks": final_responses, "total_chunks": len(chunks), "provider_info": "Fabric Chunker"},
                correlation_id=req.correlation_id
            )

        return await self._dispatch_single(req)

    # ── Dispatch singolo (con retry + circuit breaker) ───────────────────────

    async def _dispatch_single(self, req: DispatchRequest) -> DispatchResult:
        max_attempts  = 4
        last_exception: Exception = Exception("nessun tentativo eseguito")
        excluded: set[str] = set()

        for attempt in range(max_attempts):
            provider_id = req.provider_hint if attempt == 0 else None
            if provider_id is None:
                provider_id = self._select(req.capability, exclude=excluded)

            if not provider_id or provider_id not in self._specs:
                break  # nessun provider disponibile — esci dal loop

            spec  = self._specs[provider_id]
            t0    = time.time()

            try:
                response = await self._call_provider(spec, req)

                # CRIT-C: risposta con quota/rate-limit segnalato nel body
                is_quota = (
                    isinstance(response, dict)
                    and (
                        response.get("error") == "quota_or_ratelimit"
                        or response.get("status_code") in (402, 429)
                        or self._is_quota_error(str(response))
                    )
                )
                if is_quota:
                    await _token_rotator.rotate()
                    excluded.add(provider_id)
                    await asyncio.sleep(_RETRY_DELAY_S)
                    continue

                # Successo — azzera contatore errori Oracle se applicabile
                if spec.kind == ProviderKind.ORACLE:
                    self._reset_oracle_errors(provider_id)

                latency = (time.time() - t0) * 1000
                _rke("dispatch_ok", provider=provider_id, latency_ms=latency)
                return DispatchResult(
                    task_id=req.task_id, provider_id=spec.provider_id,
                    provider_name=spec.name, provider_kind=spec.kind.value,
                    capability=req.capability, status="executed",
                    response=response, latency_ms=latency,
                    correlation_id=req.correlation_id
                )

            except Exception as exc:
                last_exception = exc
                exc_str = str(exc).lower()

                # CRIT-C: pattern quota/rate-limit esteso anche nelle eccezioni
                if self._is_quota_error(exc_str):
                    await _token_rotator.rotate()
                    excluded.add(provider_id)
                    await asyncio.sleep(_RETRY_DELAY_S)
                    continue

                # CRIT-B: Oracle error → circuit breaker + fallback sul primo provider disponibile
                if spec.kind == ProviderKind.ORACLE:
                    circuit_opened = self._record_oracle_error(provider_id)
                    _logger.error(
                        "[fabric] Oracle error (attempt %d/%d): %s — circuit %s",
                        attempt + 1, max_attempts, exc,
                        "APERTO" if circuit_opened else "registrato",
                    )
                    # Cerca dinamicamente il primo provider non-Oracle non-excluded (fast-path fallback)
                    # Evita ricorsione: chiama _call_provider direttamente senza passare da dispatch()
                    fb_spec = next(
                        (s for pid, s in self._specs.items()
                         if pid not in excluded and s.kind != ProviderKind.ORACLE),
                        None,
                    )
                    if fb_spec:
                        try:
                            t1 = time.time()
                            fb_response = await self._call_provider(fb_spec, req)
                            latency = (time.time() - t1) * 1000
                            _logger.info("[fabric] Fallback Oracle→%s riuscito (%.0fms)", fb_spec.provider_id, latency)
                            return DispatchResult(
                                task_id=req.task_id, provider_id=fb_spec.provider_id,
                                provider_name=fb_spec.name, provider_kind=fb_spec.kind.value,
                                capability=req.capability, status="executed",
                                response=fb_response, latency_ms=latency,
                                correlation_id=req.correlation_id
                            )
                        except Exception as fb_exc:
                            _logger.error("[fabric] Fallback Oracle→%s fallito: %s", fb_spec.provider_id, fb_exc)
                            last_exception = fb_exc
                    else:
                        _logger.warning("[fabric] Nessun provider fallback non-Oracle disponibile")

                self._update_health(provider_id, ProviderHealth.DOWN)
                excluded.add(provider_id)

        return DispatchResult(
            task_id=req.task_id, provider_id="failed", provider_name="none",
            provider_kind="none", capability=req.capability, status="failed",
            response={"error": str(last_exception)},
            correlation_id=req.correlation_id
        )