server/curriculum.py

"""
Curriculum manager — auto-advances difficulty based on rolling success rate.

easy → medium → hard, with retreat if performance drops.
The environment calls update() after each episode and reads current_config().
"""
from .config import EnvConfig, CurriculumConfig, RealismConfig, TaskConfig


_LEVELS = {
    "easy":   EnvConfig.easy(),
    "medium": EnvConfig.medium(),
    "hard":   EnvConfig.hard(),
}


class CurriculumManager:
    def __init__(self, cfg: CurriculumConfig):
        self._cfg = cfg
        self._level_idx = 0
        self._levels = cfg.levels
        self._advance_count = 0   # consecutive episodes above threshold
        self._retreat_count = 0

    @property
    def current_level(self) -> str:
        return self._levels[self._level_idx]

    def current_config(self) -> EnvConfig:
        return _LEVELS[self.current_level]

    def update(self, success_rate: float) -> str:
        """Call after each episode. Returns level name (may have changed)."""
        if not self._cfg.enabled:
            return self.current_level

        if success_rate >= self._cfg.advance_threshold:
            self._advance_count += 1
            self._retreat_count = 0
            if self._advance_count >= 5 and self._level_idx < len(self._levels) - 1:
                self._level_idx += 1
                self._advance_count = 0
                print(f"[Curriculum] Advanced to {self.current_level} "
                      f"(success_rate={success_rate:.0%})")
        elif success_rate <= self._cfg.retreat_threshold:
            self._retreat_count += 1
            self._advance_count = 0
            if self._retreat_count >= 3 and self._level_idx > 0:
                self._level_idx -= 1
                self._retreat_count = 0
                print(f"[Curriculum] Retreated to {self.current_level} "
                      f"(success_rate={success_rate:.0%})")
        else:
            self._advance_count = 0
            self._retreat_count = 0

        return self.current_level