Spaces:
Running
Running
| 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"^(?P<major>0|[1-9]\d*)\.(?P<minor>0|[1-9]\d*)\.(?P<patch>0|[1-9]\d*)" | |
| rf"(?:-(?P<prerelease>{_SEMVER_IDENTIFIER}(?:\.{_SEMVER_IDENTIFIER})*))?" | |
| rf"(?:\+(?P<build>[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 | |
| 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. | |
| """ | |
| 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() | |