"""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", ]