""" backend/api/capability_resolver.py — Capability Resolver (ARCH-E3.2) Mappa le capability richieste dal Brain ai Worker disponibili nel Capability Catalog (ARCH-E3.1), scegliendo il provider ottimale in base a SLA, GPU, regione e salute operativa. Posizione nel flusso: Brain → Kernel.submit_task(capability) → Resolver.resolve() → provider_hint → ExecutionFabric.dispatch(provider_hint) → Worker Differenza con ExecutionFabric._select(): - Resolver: decisione DICHIARATIVA dal Catalog (metadata statici, SLA contratto) - Fabric._select(): decisione OPERATIVA (health live, circuit breaker, concurrency) Il resolver fornisce l'hint; il Fabric può ignorarlo se il provider è down. Funzionalità: resolve(ResolveRequest) → ResolveResult (migliore provider + alternative) resolve_many([ResolveReq]) → list[ResolveResult] (bulk per Workflow Engine) can_resolve(capability) → bool (quick check senza scoring) HTTP Endpoints (auth: MACHINE): POST /api/resolver/resolve — risolve una singola capability POST /api/resolver/resolve-many — risolve N capability in bulk (workflow planning) GET /api/resolver/status — diagnostica: capabilities risolvibili, contatori Invarianti ADR: S4: Brain non conosce l'infrastruttura S9: ogni servizio ignora l'impl interna degli altri S19: nessun vendor lock-in — chiunque nel Catalog è eleggibile S20: routing intent-based, non hardcoded S27: ogni risoluzione tracciata via resolve_id """ from __future__ import annotations import logging import time import uuid from typing import Any from fastapi import APIRouter, Depends, HTTPException from pydantic import BaseModel, Field from .auth_guard import AuthRole, require_role _logger = logging.getLogger("api.capability_resolver") # ── Catalog import (guard) ───────────────────────────────────────────────────── try: from .capability_catalog import catalog as _catalog, CapabilityDescriptor as _CapDesc _CATALOG_AVAILABLE = True except Exception: _catalog = None # type: ignore[assignment] _CapDesc = None # type: ignore[assignment] _CATALOG_AVAILABLE = False # ── Fabric state import (guard) — per leggere health live senza accoppiamento ── try: from .execution_fabric import fabric as _fabric _FABRIC_AVAILABLE = True except Exception: _fabric = None # type: ignore[assignment] _FABRIC_AVAILABLE = False # ── Scoring weights ──────────────────────────────────────────────────────────── # ARCH-RESOLVER-FB: aggiunto _W_FEEDBACK — pesi bilanciati a 1.00 _W_SLA = 0.35 # peso SLA target (latenza dichiarata) _W_COST = 0.15 # peso costo (free > paid) _W_ALWAYS_ON = 0.20 # peso always-on vs on-demand _W_REGION = 0.15 # peso preferenza regione _W_FEEDBACK = 0.15 # peso feedback storico real-world (success rate + latency delta) # ── Feedback Tracker — EWA in-memory, zero I/O ──────────────────────────────── class _FeedbackRecord: """Record EWA (Exponentially Weighted Average) per provider.""" __slots__ = ("success_ewa", "latency_ratio_ewa", "calls", "_alpha") def __init__(self, alpha: float = 0.2) -> None: self.success_ewa = 1.0 # parte ottimista (assume ok fino a prova contraria) self.latency_ratio_ewa = 1.0 # actual_ms / declared_sla_ms (1.0 = rispetta SLA) self.calls = 0 self._alpha = alpha def record(self, success: bool, actual_ms: float | None, declared_sla_ms: float) -> None: a = self._alpha self.success_ewa = (1 - a) * self.success_ewa + a * (1.0 if success else 0.0) if actual_ms is not None and declared_sla_ms > 0: ratio = actual_ms / declared_sla_ms self.latency_ratio_ewa = (1 - a) * self.latency_ratio_ewa + a * ratio self.calls += 1 def score(self) -> float: """Score [0,1]: 1.0 = perfetto (success rate 100%, rispetta SLA), 0 = pessimo.""" # success rate: 1.0 → bonus, 0.0 → forte penalità s_score = self.success_ewa # latency ratio: ratio ≤ 1 (batte SLA) → bonus, ratio > 2 → forte penalità l_score = min(1.0, 1.0 / max(self.latency_ratio_ewa, 0.5)) return 0.7 * s_score + 0.3 * l_score class FeedbackTracker: """Registry in-memory di feedback per provider_id. Thread-safe tramite GIL.""" def __init__(self) -> None: self._records: dict[str, _FeedbackRecord] = {} def record(self, provider_id: str, success: bool, actual_ms: float | None = None, declared_sla_ms: float = 1000.0) -> None: if provider_id not in self._records: self._records[provider_id] = _FeedbackRecord() self._records[provider_id].record(success, actual_ms, declared_sla_ms) def score(self, provider_id: str) -> float: """Restituisce feedback score [0,1]. Default ottimistico 0.85 se nessun dato.""" rec = self._records.get(provider_id) return rec.score() if rec else 0.85 def stats(self) -> dict: return { pid: {"calls": r.calls, "success_ewa": round(r.success_ewa, 3), "latency_ratio_ewa": round(r.latency_ratio_ewa, 3), "score": round(r.score(), 3)} for pid, r in self._records.items() } # ── Models ───────────────────────────────────────────────────────────────────── class ResolveRequest(BaseModel): """Richiesta di risoluzione capability da parte del Brain (via Kernel).""" capability: str = Field(..., description="Nome capability richiesta, es. 'browser'") require_gpu: bool = Field(False) max_sla_ms: float | None = Field(None, description="SLA massimo accettato in ms") prefer_region: str | None = Field(None, description="Regione preferita, es. 'eu'") tags: list[str] = Field(default_factory=list, description="Tag intent-based extra") payload_kb: int = Field(0, description="Stima dimensione payload in KB") exclude_providers: list[str] = Field(default_factory=list, description="Provider da escludere") correlation_id: str = Field(default_factory=lambda: str(uuid.uuid4())) class ProviderCandidate(BaseModel): """Provider candidato per una capability con score e metadati.""" provider_id: str provider_name: str sla_ms: float version: str cost_unit: float region: str always_on: bool requires_gpu: bool score: float = Field(description="Score composito [0,1]") tags: list[str] = Field(default_factory=list) class ResolveResult(BaseModel): """Risultato della risoluzione — provider ottimale + alternative ordinate.""" resolve_id: str = Field(default_factory=lambda: str(uuid.uuid4())) capability: str resolved: bool = False provider_id: str | None = None provider_name: str | None = None sla_ms: float | None = None version: str = "1.0.0" score: float = 0.0 alternatives: list[ProviderCandidate] = Field(default_factory=list) reason: str = "" resolved_at: float = Field(default_factory=time.time) catalog_source: bool = True # True = da Catalog, False = fallback fabric class ResolveManyRequest(BaseModel): requests: list[ResolveRequest] = Field(..., description="Lista richieste da risolvere in bulk") # ── CapabilityResolver singleton ─────────────────────────────────────────────── class CapabilityResolver: """ Risolve capability → provider ottimale usando il Capability Catalog. Algoritmo (puramente dichiarativo, non modifica stato): 1. Query catalog per capability (+ filtri hard: GPU, SLA, payload_kb) 2. Per ogni candidato, calcola score composito: score = W_SLA * sla_score + W_COST * cost_score + W_ALWAYS_ON * aon_score + W_REGION * region_score 3. Ordina candidati per score desc 4. Ritorna best + ordered alternatives Se il Catalog non è disponibile o vuoto, tenta fallback sul Fabric (usa _fabric._specs per lista provider registrati). """ # ── Resolve (singola) ───────────────────────────────────────────────────── def resolve(self, req: ResolveRequest) -> ResolveResult: """ Risoluzione sincrona — il Catalog è un dict in-memory, nessuna I/O. Chiamabile sia da codice sync che async. """ rid = str(uuid.uuid4()) candidates = self._query_candidates(req) if not candidates: # Fallback: prova dal Fabric se il Catalog è vuoto candidates = self._fallback_from_fabric(req) if not candidates: _logger.warning("[resolver] no provider for capability=%s", req.capability) return ResolveResult( resolve_id=rid, capability=req.capability, resolved=False, reason=f"Nessun provider disponibile per capability '{req.capability}'", ) scored = sorted(candidates, key=lambda c: c.score, reverse=True) best = scored[0] alts = scored[1:] _logger.info("[resolver] resolved cap=%s → provider=%s sla=%.0fms score=%.3f alts=%d", req.capability, best.provider_id, best.sla_ms, best.score, len(alts)) return ResolveResult( resolve_id = rid, capability = req.capability, resolved = True, provider_id = best.provider_id, provider_name = best.provider_name, sla_ms = best.sla_ms, version = best.version, score = best.score, alternatives = alts[:5], # max 5 alternative reason = "ok", ) # ── Resolve Many (bulk, per Workflow Engine) ─────────────────────────────── def resolve_many(self, requests: list[ResolveRequest]) -> list[ResolveResult]: """Risolve N capability in bulk. Usato dal Workflow Engine (ARCH-I4.2).""" return [self.resolve(r) for r in requests] # ── can_resolve (quick check) ───────────────────────────────────────────── def can_resolve(self, capability: str) -> bool: """Ritorna True se esiste almeno un provider vivo per questa capability.""" if _CATALOG_AVAILABLE and _catalog is not None: return len(_catalog.query(name=capability)) > 0 if _FABRIC_AVAILABLE and _fabric is not None: return any( capability in spec.capabilities for spec in _fabric._specs.values() if spec.base_url ) return False # ── Internal: query candidates ──────────────────────────────────────────── def _query_candidates(self, req: ResolveRequest) -> list[ProviderCandidate]: if not _CATALOG_AVAILABLE or _catalog is None: return [] entries = _catalog.query( name = req.capability, requires_gpu = req.require_gpu or None, # None = non filtrare max_sla_ms = req.max_sla_ms, region = None, # regione usata solo per scoring, non filtro hard ) candidates = [] for e in entries: # Filtri hard addizionali if req.require_gpu and not e.requires_gpu: continue if e.provider_id in req.exclude_providers: continue if req.payload_kb and req.payload_kb > e.max_payload_kb: continue score = self._score(e, req) candidates.append(ProviderCandidate( provider_id = e.provider_id, provider_name = e.provider_name, sla_ms = e.sla_ms, version = e.version, cost_unit = e.cost_unit, region = e.region, always_on = e.always_on, requires_gpu = e.requires_gpu, tags = e.tags, score = score, )) return candidates def _fallback_from_fabric(self, req: ResolveRequest) -> list[ProviderCandidate]: """ Fallback: legge _specs dal Fabric se il Catalog è vuoto o non disponibile. Usato solo quando il Fabric non ha ancora fatto initialize() + auto-register. """ if not _FABRIC_AVAILABLE or _fabric is None: return [] candidates = [] for pid, spec in _fabric._specs.items(): if req.capability not in spec.capabilities: continue if req.require_gpu and not spec.gpu: continue if pid in req.exclude_providers: continue if not spec.base_url: continue sla = 9000.0 # default conservativo cost = spec.cost_unit always_on = hasattr(spec, 'always_on') and str(spec.always_on) not in ("on-demand", "no") score = self._score_raw(sla, cost, always_on, spec.region, req.prefer_region) candidates.append(ProviderCandidate( provider_id = spec.provider_id, provider_name = spec.name, sla_ms = sla, version = "1.0.0", cost_unit = cost, region = spec.region, always_on = always_on, requires_gpu = spec.gpu, score = score, )) return candidates # ── Scoring ─────────────────────────────────────────────────────────────── def _score(self, e: "_CapDesc", req: ResolveRequest) -> float: # type: ignore[name-defined] always_on = e.always_on feedback_score = self._feedback.score(e.name) # e.name = provider_id nel catalog return self._score_raw(e.sla_ms, e.cost_unit, always_on, e.region, req.prefer_region, feedback_score) @staticmethod def _score_raw(sla_ms: float, cost: float, always_on: bool, region: str, prefer_region: str | None, feedback_score: float = 0.85) -> float: # SLA score: SLA bassa → score alto. Riferimento 5000ms. sla_score = min(1.0, 5000.0 / max(sla_ms, 100.0)) # Cost score: free → 1.0, 1 unit → 0.5 cost_score = 1.0 / (1.0 + cost * 10) # Always-on score aon_score = 1.0 if always_on else 0.4 # Region score region_score = 1.0 if (not prefer_region or region == prefer_region) else 0.7 # Feedback score: EWA di success rate + latency ratio reale (ARCH-RESOLVER-FB) return ( _W_SLA * sla_score + _W_COST * cost_score + _W_ALWAYS_ON * aon_score + _W_REGION * region_score + _W_FEEDBACK * feedback_score ) # ── Feedback recording ──────────────────────────────────────────────────── def record_feedback(self, provider_id: str, success: bool, actual_ms: float | None = None, declared_sla_ms: float = 1000.0) -> None: """ Registra il risultato reale di una chiamata al provider (ARCH-RESOLVER-FB). Chiamato dall'Executor/Brain dopo ogni tool execution. """ self._feedback.record(provider_id, success, actual_ms, declared_sla_ms) _logger.debug("[resolver] feedback %s → success=%s actual_ms=%s", provider_id, success, actual_ms) # ── Status ──────────────────────────────────────────────────────────────── def status(self) -> dict: resolvable: list[str] = [] if _CATALOG_AVAILABLE and _catalog is not None: entries = _catalog.all_entries() resolvable = sorted({e.name for e in entries}) return { "catalog_available": _CATALOG_AVAILABLE, "fabric_available": _FABRIC_AVAILABLE, "resolvable_capabilities": resolvable, "total_resolvable": len(resolvable), "weights": { "sla": _W_SLA, "cost": _W_COST, "always_on": _W_ALWAYS_ON, "region": _W_REGION, "feedback": _W_FEEDBACK, }, "feedback_stats": self._feedback.stats(), } # ── Singleton ─────────────────────────────────────────────────────────────────── resolver = CapabilityResolver() # ── HTTP Router ───────────────────────────────────────────────────────────────── router = APIRouter( prefix="/api/resolver", tags=["capability-resolver"], dependencies=[Depends(require_role(AuthRole.MACHINE))], ) @router.post("/resolve", summary="Risolve una capability → provider ottimale") async def route_resolve(req: ResolveRequest) -> ResolveResult: result = resolver.resolve(req) if not result.resolved: raise HTTPException(404, result.reason) return result @router.post("/resolve-many", summary="Risolve N capability in bulk (workflow planning)") async def route_resolve_many(req: ResolveManyRequest) -> dict: if not req.requests: raise HTTPException(400, "requests lista vuota") results = resolver.resolve_many(req.requests) resolved = sum(1 for r in results if r.resolved) unresolved = len(results) - resolved return { "total": len(results), "resolved": resolved, "unresolved": unresolved, "results": [r.model_dump() for r in results], } @router.get("/status", summary="Stato resolver — capabilities risolvibili e pesi scoring") async def route_status() -> dict: return resolver.status() class ResolverFeedback(BaseModel): """Feedback da inviare dopo l'esecuzione di un tool (ARCH-RESOLVER-FB).""" provider_id: str success: bool actual_ms: float | None = None declared_sla_ms: float = 1000.0 @router.post("/feedback", summary="Registra feedback reale su un provider (latenza, successo)") async def route_feedback(body: ResolverFeedback) -> dict: """ Chiamato dall'Executor dopo ogni tool execution per aggiornare lo scoring EWA. Non-critico: un errore qui non deve mai bloccare l'esecuzione dell'agente. """ resolver.record_feedback(body.provider_id, body.success, body.actual_ms, body.declared_sla_ms) return {"recorded": True, "provider_id": body.provider_id}