praxis/server/scenarios/library.py

"""Scenario library — index manifest + on-demand loader (SLICE-02, REQ-SCEN-03).

Loads scenarios/index.yaml (a slim manifest), then loads individual scenario
YAMLs on demand via server/scenarios/loader.py and validates them against the
Pydantic schema. Provides IRT-aware selection (select_for_theta) and a CI-
checkable coverage method (check_coverage) enforcing MIN_COVERAGE = 2 scenarios
per rubric criterion (RESEARCH §D).
"""

from __future__ import annotations

import math
import re
from pathlib import Path

import yaml
from pydantic import BaseModel, Field, ValidationError, field_validator

from server.scenarios.loader import load as load_scenario
from server.scenarios.schema import Scenario

_DEFAULT_SCENARIOS_DIR = Path(__file__).resolve().parent.parent.parent / "scenarios"
_SEMVER_RE = re.compile(r"^\d+\.\d+\.\d+(?:-[0-9A-Za-z.-]+)?(?:\+[0-9A-Za-z.-]+)?$")


class IndexEntry(BaseModel):
    """One row in scenarios/index.yaml."""

    id: str = Field(..., description="Scenario id (matches the scenario YAML id field)")
    path: str = Field(..., description="Relative path to the scenario YAML from scenarios/")
    title: str
    difficulty: int = Field(..., ge=1, le=5)
    failure_mode: str
    rubric_criteria: list[str] = Field(default_factory=list)
    version: str = Field("1.0.0")
    author: str = Field("expert")
    generated_from: str | None = None

    @field_validator("version")
    @classmethod
    def _validate_semver(cls, v: str) -> str:
        if not _SEMVER_RE.match(v):
            raise ValueError(f"invalid semver: {v!r}")
        return v


class IndexManifest(BaseModel):
    version: str = Field("1.0.0")
    scenarios: list[IndexEntry] = Field(default_factory=list)

    @field_validator("version")
    @classmethod
    def _validate_semver(cls, v: str) -> str:
        if not _SEMVER_RE.match(v):
            raise ValueError(f"invalid semver: {v!r}")
        return v


class CoverageError(Exception):
    """Raised when a rubric criterion has fewer than MIN_COVERAGE scenarios."""


def _logit(p: float) -> float:
    return math.log(p / (1.0 - p))


class ScenarioLibrary:
    """Loads scenarios/index.yaml and serves scenarios on demand.

    Lazy: the manifest is loaded once; individual scenario YAMLs are parsed
    on first get() and cached.
    """

    MIN_COVERAGE = 2

    def __init__(self, scenarios_dir: Path | None = None) -> None:
        self.scenarios_dir = scenarios_dir or _DEFAULT_SCENARIOS_DIR
        self._index_path = self.scenarios_dir / "index.yaml"
        self._manifest: IndexManifest | None = None
        self._cache: dict[str, Scenario] = {}

    def load(self) -> IndexManifest:
        """Load and validate the index manifest. Idempotent."""
        if self._manifest is not None:
            return self._manifest
        if not self._index_path.exists():
            raise FileNotFoundError(f"Scenario index not found: {self._index_path}")
        with self._index_path.open("r", encoding="utf-8") as f:
            raw = yaml.safe_load(f)
        self._manifest = IndexManifest.model_validate(raw)
        return self._manifest

    @property
    def manifest(self) -> IndexManifest:
        if self._manifest is None:
            self.load()
        assert self._manifest is not None
        return self._manifest

    def entries(self) -> list[IndexEntry]:
        return list(self.manifest.scenarios)

    def get(self, scenario_id: str) -> Scenario:
        """Load (and cache) a scenario by id, validating against the schema."""
        if scenario_id in self._cache:
            return self._cache[scenario_id]
        entry = self._entry_by_id(scenario_id)
        scenario = load_scenario(entry.id, scenarios_dir=self.scenarios_dir)
        if scenario.id != entry.id:
            raise ValueError(
                f"index/scenario id mismatch: index={entry.id!r} yaml={scenario.id!r}"
            )
        if scenario.version != entry.version:
            raise ValueError(
                f"version mismatch for {scenario_id}: index={entry.version!r} yaml={scenario.version!r}"
            )
        self._cache[scenario_id] = scenario
        return scenario

    def _entry_by_id(self, scenario_id: str) -> IndexEntry:
        for e in self.manifest.scenarios:
            if e.id == scenario_id:
                return e
        raise KeyError(f"scenario id not in index: {scenario_id}")

    def list_by_path(self, path: str) -> list[IndexEntry]:
        """List index entries whose scenario.path matches the given skill path."""
        out: list[IndexEntry] = []
        for e in self.manifest.scenarios:
            s = self.get(e.id)
            if s.path == path:
                out.append(e)
        return out

    def list_by_difficulty(self, min_difficulty: int, max_difficulty: int) -> list[IndexEntry]:
        """List index entries with difficulty in [min, max] inclusive."""
        out: list[IndexEntry] = []
        for e in self.manifest.scenarios:
            if min_difficulty <= e.difficulty <= max_difficulty:
                out.append(e)
        return out

    def select_for_theta(
        self, theta: float, path: str, target_p: float = 0.7
    ) -> Scenario | None:
        """IRT-aware scenario selection.

        Picks the scenario (within the given path) whose difficulty b is
        closest to theta - logit(target_p), so that the predicted P_success
        is near target_p. Returns None if the path has no scenarios.

        Per SLICE-02/TASK-02-03 and the IRT selection formula
        (b* = theta - logit(p); logit(p) = ln(p/(1-p))).
        """
        entries = self.list_by_path(path)
        if not entries:
            return None
        target_b = theta - _logit(target_p)
        best_entry: IndexEntry | None = None
        best_dist = math.inf
        for e in entries:
            dist = abs(float(e.difficulty) - target_b)
            if dist < best_dist:
                best_dist = dist
                best_entry = e
        assert best_entry is not None
        return self.get(best_entry.id)

    def check_coverage(self, path: str) -> dict[str, int]:
        """Verify each rubric criterion in the path has >= MIN_COVERAGE scenarios.

        Returns a {criterion_id: scenario_count} map. Raises CoverageError if
        any criterion is under-covered. CI-callable.
        """
        entries = self.list_by_path(path)
        counts: dict[str, int] = {}
        for e in entries:
            for cid in e.rubric_criteria:
                counts[cid] = counts.get(cid, 0) + 1
        under = {cid: n for cid, n in counts.items() if n < self.MIN_COVERAGE}
        if under:
            raise CoverageError(
                f"rubric criteria under MIN_COVERAGE={self.MIN_COVERAGE} for path {path!r}: {under}"
            )
        return counts


__all__ = [
    "ScenarioLibrary",
    "IndexEntry",
    "IndexManifest",
    "CoverageError",
    "ValidationError",
]