api/priority.py

"""
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),
    }