""" backend/api/priority.py — Priority job semaphores (S-DUAL-1) Due classi di job con concorrenza controllata via asyncio.Semaphore: REALTIME — agent steps interattivi, exec code da UI, terminal commands Semaphore(6): latency-sensitive, risposta attesa < 30s BACKGROUND — benchmark, research multi-URL, pip-install headless Semaphore(2): best-effort, può attendere, non blocca mai REALTIME I contatori live sono esposti da /api/health/load per il routing adattivo CF. """ import asyncio, time, logging from contextlib import asynccontextmanager from typing import AsyncGenerator _logger = logging.getLogger("api.priority") _boot_time = time.monotonic() # ── Semaphores ───────────────────────────────────────────────────────────────── _REALTIME_LIMIT = 6 _BACKGROUND_LIMIT = 2 _realtime_sem = asyncio.Semaphore(_REALTIME_LIMIT) _background_sem = asyncio.Semaphore(_BACKGROUND_LIMIT) # Contatori atomici per metriche /api/health/load _realtime_active = 0 _background_active = 0 @asynccontextmanager async def realtime_job(timeout_s: float = 300.0) -> AsyncGenerator[None, None]: """ Context manager per job REALTIME (agent steps, exec interattivo, terminal). Acquisisce il semaphore con timeout — rilancia asyncio.TimeoutError se non ci sono slot liberi entro timeout_s (default 300s = non dovrebbe mai scadere per richieste UI normali, ma protegge da leak di semaphore). Uso: async with realtime_job(): result = await run_subprocess(...) """ global _realtime_active try: await asyncio.wait_for(_realtime_sem.acquire(), timeout=timeout_s) except asyncio.TimeoutError: _logger.warning("[priority] REALTIME semaphore timeout dopo %.0fs", timeout_s) raise _realtime_active += 1 try: yield finally: _realtime_active = max(0, _realtime_active - 1) _realtime_sem.release() @asynccontextmanager async def background_job(timeout_s: float = 30.0) -> AsyncGenerator[None, None]: """ Context manager per job BACKGROUND (benchmark, research, pip-install). Timeout più aggressivo (default 30s): se entrambi gli slot BACKGROUND sono occupati e non si liberano in 30s → 429 Too Many Requests al chiamante. Garantisce che il benchmark non blocchi mai i job REALTIME. Uso: async with background_job(timeout_s=30.0): result = await run_benchmark(...) """ global _background_active try: await asyncio.wait_for(_background_sem.acquire(), timeout=timeout_s) except asyncio.TimeoutError: _logger.warning("[priority] BACKGROUND semaphore timeout dopo %.0fs — job rifiutato", timeout_s) raise _background_active += 1 try: yield finally: _background_active = max(0, _background_active - 1) _background_sem.release() def get_load_metrics() -> dict: """ Metriche live per /api/health/load. realtime_waiting: slot REALTIME occupati (Semaphore usa valore interno). Il valore _sem._value è il numero di slot LIBERI. """ return { "realtime_active": _realtime_active, "realtime_capacity": _REALTIME_LIMIT, "realtime_available": _realtime_sem._value, # slot liberi "background_active": _background_active, "background_capacity": _BACKGROUND_LIMIT, "background_available": _background_sem._value, # slot liberi "uptime_s": int(time.monotonic() - _boot_time), }