import logging import re import time from typing import Optional, Tuple from pydantic import BaseModel, field_validator from .marketplace import WORKERS_REGISTRY, WorkerCapability _logger = logging.getLogger("api.resolver") _SEMVER_IDENTIFIER = r"(?:0|[1-9]\d*|\d*[A-Za-z-][0-9A-Za-z-]*)" _SEMVER_PATTERN = re.compile( rf"^(?P0|[1-9]\d*)\.(?P0|[1-9]\d*)\.(?P0|[1-9]\d*)" rf"(?:-(?P{_SEMVER_IDENTIFIER}(?:\.{_SEMVER_IDENTIFIER})*))?" rf"(?:\+(?P[0-9A-Za-z-]+(?:\.[0-9A-Za-z-]+)*))?$" ) def _parse_semver(version: str) -> Tuple[int, int, int, Optional[Tuple[str, ...]]]: """Parsa una versione Semantic Versioning 2.0.0 senza dipendenze esterne.""" match = _SEMVER_PATTERN.fullmatch(version) if not match: raise ValueError(f"Invalid Semantic Version: {version!r}") prerelease = match.group("prerelease") return ( int(match.group("major")), int(match.group("minor")), int(match.group("patch")), tuple(prerelease.split(".")) if prerelease else None, ) def _compare_semver(left: str, right: str) -> int: """Confronta due versioni SemVer, restituendo -1, 0 oppure 1.""" left_major, left_minor, left_patch, left_prerelease = _parse_semver(left) right_major, right_minor, right_patch, right_prerelease = _parse_semver(right) left_core = (left_major, left_minor, left_patch) right_core = (right_major, right_minor, right_patch) if left_core != right_core: return -1 if left_core < right_core else 1 if left_prerelease is None and right_prerelease is None: return 0 if left_prerelease is None: return 1 if right_prerelease is None: return -1 for left_identifier, right_identifier in zip(left_prerelease, right_prerelease): if left_identifier == right_identifier: continue left_is_numeric = left_identifier.isdigit() right_is_numeric = right_identifier.isdigit() if left_is_numeric and right_is_numeric: return -1 if int(left_identifier) < int(right_identifier) else 1 if left_is_numeric != right_is_numeric: return -1 if left_is_numeric else 1 return -1 if left_identifier < right_identifier else 1 if len(left_prerelease) == len(right_prerelease): return 0 return -1 if len(left_prerelease) < len(right_prerelease) else 1 class ResolverConstraints(BaseModel): min_version: Optional[str] = None max_cost: Optional[float] = None max_latency: Optional[float] = None preferred_region: Optional[str] = None require_gpu: bool = False min_priority: int = 100 @field_validator("min_version") @classmethod def validate_min_version(cls, value: Optional[str]) -> Optional[str]: if value is not None: _parse_semver(value) return value class CapabilityResolver: """ ARCH-E3.2: Capability Resolver Mappa le capacità richieste dal Brain ai Worker disponibili tramite il Marketplace, scegliendo il migliore in base agli SLA. """ @staticmethod async def resolve( capability: str, constraints: Optional[ResolverConstraints] = None, ) -> Optional[WorkerCapability]: """ Risolve una capability in un Worker specifico. Strategia: 1. Filtra per capability supportata. 2. Filtra per worker attivi (last_seen < 300s). 3. Applica constraints (versione, costo, latenza, GPU, priorità). 4. Se disponibile, preferisce la regione richiesta. 5. Ordina per (priority ASC, cost ASC, latency ASC). """ now = int(time.time()) candidates = [] from .health_manager import health_manager for worker in WORKERS_REGISTRY.values(): # 1. & 2. Filtro base + Health Check (ARCH-P5.1) is_alive = now - worker.last_seen < 300 is_healthy = await health_manager.is_healthy(worker.id) if capability not in worker.capabilities or not is_alive or not is_healthy: continue # 3. Applica constraints if constraints: if constraints.min_version: try: if _compare_semver(worker.version, constraints.min_version) < 0: continue except ValueError: _logger.warning( "Worker %s escluso: versione non valida per il vincolo SemVer (%r)", worker.id, worker.version, ) continue if constraints.max_cost is not None and worker.cost > constraints.max_cost: continue if constraints.max_latency is not None and worker.latency > constraints.max_latency: continue if constraints.require_gpu and not worker.gpu: continue # Nel Marketplace una priorità più bassa è migliore; min_priority # mantiene il nome del contratto esistente come soglia massima accettata. if worker.priority > constraints.min_priority: continue candidates.append(worker) if constraints and constraints.preferred_region: regional_candidates = [ worker for worker in candidates if worker.region == constraints.preferred_region ] if regional_candidates: candidates = regional_candidates if not candidates: _logger.warning(f"Nessun worker trovato per capability: {capability}") return None # 5. Ordinamento per SLA # Priorità: Priority (basso meglio), Cost (basso meglio), Latency (basso meglio) candidates.sort(key=lambda worker: (worker.priority, worker.cost, worker.latency)) best_worker = candidates[0] _logger.info( "Risolta capability '%s' su worker '%s' (score: p=%s, c=%s, l=%s)", capability, best_worker.id, best_worker.priority, best_worker.cost, best_worker.latency, ) return best_worker # Singleton instance resolver = CapabilityResolver()