Terminal / api /resolver.py
Baida07's picture
sync: 175 file da Baida98/AI@d1881b9c (2026-08-25 07:39 UTC) [deploy-all] (#67)
bd654eb
Raw
History Blame
6.38 kB
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()