Spaces:
Running
Running
File size: 6,383 Bytes
28a08e7 bd654eb 28a08e7 bd654eb 28a08e7 bd654eb 28a08e7 bd654eb 28a08e7 bd654eb 28a08e7 bd654eb 28a08e7 bd654eb 28a08e7 bd654eb 28a08e7 bd654eb 28a08e7 bd654eb 28a08e7 bd654eb 28a08e7 bd654eb 28a08e7 bd654eb 28a08e7 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 | 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
@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()
|