api/priority.py

"""
backend/api/priority.py — Priority job semaphores (S-DUAL-1)

Definisce classi di job con concorrenza controllata via asyncio.Semaphore:
  HIGH       — agent steps interattivi, exec code da UI, terminal commands
               Semaphore(6): latency-sensitive, risposta attesa < 30s
  NORMAL     — task di media priorità, elaborazioni non critiche
               Semaphore(4): bilanciamento tra latenza e throughput
  LOW        — task a bassa priorità, come pre-calcoli o aggiornamenti in background
               Semaphore(2): può attendere, non blocca job più importanti
  BACKGROUND — benchmark, research multi-URL, pip-install headless
               Semaphore(2): best-effort, può attendere, non blocca mai HIGH/NORMAL/LOW

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 ─────────────────────────────────────────────────────────────────
_HIGH_LIMIT        = 6
_NORMAL_LIMIT      = 4
_LOW_LIMIT         = 2
_BACKGROUND_LIMIT  = 2

_high_sem        = asyncio.Semaphore(_HIGH_LIMIT)
_normal_sem      = asyncio.Semaphore(_NORMAL_LIMIT)
_low_sem         = asyncio.Semaphore(_LOW_LIMIT)
_background_sem  = asyncio.Semaphore(_BACKGROUND_LIMIT)

# Contatori atomici per metriche /api/health/load
_high_active        = 0
_normal_active      = 0
_low_active         = 0
_background_active  = 0


@asynccontextmanager
async def high_priority_job(timeout_s: float = 300.0) -> AsyncGenerator[None, None]:
    """
    Context manager per job HIGH (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 high_priority_job():
            result = await run_subprocess(...)
    """
    global _high_active
    try:
        await asyncio.wait_for(_high_sem.acquire(), timeout=timeout_s)
    except asyncio.TimeoutError:
        _logger.warning("[priority] HIGH priority semaphore timeout dopo %.0fs", timeout_s)
        raise

    _high_active += 1
    try:
        yield
    finally:
        _high_active = max(0, _high_active - 1)
        _high_sem.release()


@asynccontextmanager
async def normal_priority_job(timeout_s: float = 120.0) -> AsyncGenerator[None, None]:
    """
    Context manager per job NORMAL (task di media priorità).
    Uso:
        async with normal_priority_job():
            result = await process_data(...)
    """
    global _normal_active
    try:
        await asyncio.wait_for(_normal_sem.acquire(), timeout=timeout_s)
    except asyncio.TimeoutError:
        _logger.warning("[priority] NORMAL priority semaphore timeout dopo %.0fs", timeout_s)
        raise

    _normal_active += 1
    try:
        yield
    finally:
        _normal_active = max(0, _normal_active - 1)
        _normal_sem.release()


@asynccontextmanager
async def low_priority_job(timeout_s: float = 60.0) -> AsyncGenerator[None, None]:
    """
    Context manager per job LOW (task a bassa priorità).
    Uso:
        async with low_priority_job():
            result = await update_cache(...)
    """
    global _low_active
    try:
        await asyncio.wait_for(_low_sem.acquire(), timeout=timeout_s)
    except asyncio.TimeoutError:
        _logger.warning("[priority] LOW priority semaphore timeout dopo %.0fs", timeout_s)
        raise

    _low_active += 1
    try:
        yield
    finally:
        _low_active = max(0, _low_active - 1)
        _low_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.
    """
    return {
        "high_active":         _high_active,
        "high_capacity":       _HIGH_LIMIT,
        "high_available":      _high_sem._value,
        "normal_active":       _normal_active,
        "normal_capacity":     _NORMAL_LIMIT,
        "normal_available":    _normal_sem._value,
        "low_active":          _low_active,
        "low_capacity":        _LOW_LIMIT,
        "low_available":       _low_sem._value,
        "background_active":   _background_active,
        "background_capacity": _BACKGROUND_LIMIT,
        "background_available": _background_sem._value,
        "uptime_s":            int(time.monotonic() - _boot_time),
    }


# Mapping per la selezione del context manager in base alla stringa di priorità
PRIORITY_CONTEXT_MANAGERS = {
    "high": high_priority_job,
    "normal": normal_priority_job,
    "low": low_priority_job,
    "background": background_job,
}