src/runtime/resource_manager/pressure.py

"""Resource Pressure State Evaluator with Hysteresis.

Section 14:
- GREEN: Normal operation (< 70% RAM, normal queue)
- YELLOW: Reduce background work (70% - 80% RAM)
- ORANGE: Pause expensive background generation (80% - 88% RAM)
- RED: Unload non-critical models (88% - 94% RAM)
- CRITICAL: Preserve core state, pause generation, perform recovery (> 94% RAM)

Uses hysteresis thresholds to prevent rapid oscillation between states.
"""
from enum import Enum
from typing import Dict, Any


class PressureState(str, Enum):
    GREEN = "GREEN"
    YELLOW = "YELLOW"
    ORANGE = "ORANGE"
    RED = "RED"
    CRITICAL = "CRITICAL"


class PressureEvaluator:
    def __init__(self,
                 yellow_enter: float = 72.0, yellow_exit: float = 68.0,
                 orange_enter: float = 82.0, orange_exit: float = 78.0,
                 red_enter: float = 89.0, red_exit: float = 85.0,
                 critical_enter: float = 94.0, critical_exit: float = 91.0):
        self.ye_enter = yellow_enter
        self.ye_exit = yellow_exit
        self.or_enter = orange_enter
        self.or_exit = orange_exit
        self.re_enter = red_enter
        self.re_exit = red_exit
        self.cr_enter = critical_enter
        self.cr_exit = critical_exit
        self.current_state = PressureState.GREEN
        self.last_percent = 0.0

    def evaluate(self, ram_percent: float) -> PressureState:
        """Evaluates pressure state using hysteresis to avoid state flutter."""
        self.last_percent = float(ram_percent)
        
        if self.current_state == PressureState.GREEN:
            if ram_percent >= self.cr_enter:
                self.current_state = PressureState.CRITICAL
            elif ram_percent >= self.re_enter:
                self.current_state = PressureState.RED
            elif ram_percent >= self.or_enter:
                self.current_state = PressureState.ORANGE
            elif ram_percent >= self.ye_enter:
                self.current_state = PressureState.YELLOW
                
        elif self.current_state == PressureState.YELLOW:
            if ram_percent >= self.cr_enter:
                self.current_state = PressureState.CRITICAL
            elif ram_percent >= self.re_enter:
                self.current_state = PressureState.RED
            elif ram_percent >= self.or_enter:
                self.current_state = PressureState.ORANGE
            elif ram_percent <= self.ye_exit:
                self.current_state = PressureState.GREEN
                
        elif self.current_state == PressureState.ORANGE:
            if ram_percent >= self.cr_enter:
                self.current_state = PressureState.CRITICAL
            elif ram_percent >= self.re_enter:
                self.current_state = PressureState.RED
            elif ram_percent <= self.or_exit:
                self.current_state = PressureState.YELLOW
                
        elif self.current_state == PressureState.RED:
            if ram_percent >= self.cr_enter:
                self.current_state = PressureState.CRITICAL
            elif ram_percent <= self.re_exit:
                self.current_state = PressureState.ORANGE
                
        elif self.current_state == PressureState.CRITICAL:
            if ram_percent <= self.cr_exit:
                self.current_state = PressureState.RED

        return self.current_state

    def status(self) -> Dict[str, Any]:
        return {
            "state": self.current_state.value,
            "ram_percent": round(self.last_percent, 1),
            "allow_background_ai": self.current_state in (PressureState.GREEN, PressureState.YELLOW),
            "allow_speculative_work": self.current_state == PressureState.GREEN,
            "recovery_required": self.current_state == PressureState.CRITICAL
        }