feat(milestone): merge phase/01 mastery-core → milestone/v0.3-mastery-scoring

Phase 1 complete. Mastery scoring + competency rubrics + VC issuer shipped.
9 slices, 5 waves, 238 tests passing, 13/13 REQ-IDs covered.
4/4 grill MUST conditions satisfied. VERIFY: APPROVE_WITH_NOTES.

---ci---
project: praxis
phase: 1
milestone: v0.3
status: complete
requirements:
  covered: [REQ-MAST-01, REQ-MAST-02, REQ-MAST-03, REQ-SCEN-02, REQ-SCEN-03, REQ-SCEN-04, REQ-PATH-02, REQ-NFR-MAST-01, REQ-NFR-MAST-02, REQ-NFR-VC-01, REQ-NFR-VC-02, REQ-NFR-IRT-01]
  partial: []
---/ci---

server/__main__.py

@@ -32,7 +32,11 @@ from fastapi.middleware.cors import CORSMiddleware
 from fastapi.staticfiles import StaticFiles
 from pipecat.transports.smallwebrtc.connection import SmallWebRTCConnection
 
+from db.store import PraxisStore
 from server.pipeline import build_pipeline
+from server.vc.verification import verify_credential
+
+_store = PraxisStore()
 
 
 def _env(key: str, default: str = "") -> str:
@@ -117,6 +121,21 @@ async def webrtc_offer(offer: WebRTCOffer) -> dict[str, str]:
         raise HTTPException(status_code=500, detail=str(exc))
 
 
+@app.get("/vc/verify/{credential_id}")
+async def vc_verify(credential_id: str) -> dict[str, Any]:
+    """Public, unauthenticated VC verification endpoint (D-043).
+
+    Returns {valid, status, issuer, credential, mastery, credentialTier,
+    verifiedAt}. 404 if the credential id is not found. No PII beyond what
+    the credential asserts.
+    """
+    await _store.init()
+    result = await verify_credential(_store, credential_id)
+    if result is None:
+        raise HTTPException(status_code=404, detail="credential not found")
+    return result
+
+
 # ── Static client serving (D-023, REQ-DEPLOY-13) ────────────────────
 # Mount client/dist as StaticFiles at "/" AFTER all API routes so they
 # take precedence. html=True serves index.html for "/" (SPA root).

server/mastery/evidence_extractor.py

@@ -0,0 +1,204 @@
+"""Evidence extractor — LLM-extract-then-verify (SLICE-03 TASK-03-01).
+
+Off-voice-path: called after the session ends. Calls deepseek-v4-flash:cloud
+to pull verbatim-quote evidence per rubric criterion, then fuzzy-matches each
+quote against the transcript (R-MAST-02). Hallucinated quotes are rejected and
+re-extracted (max 2 attempts). On final failure the scenario is marked
+`scoring_inconclusive=True` — it does NOT silently fail to zero and does NOT
+penalize the learner (grill Axis 4 MUST #3).
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+from difflib import SequenceMatcher
+from typing import Any
+
+from pydantic import BaseModel, Field, ValidationError
+
+from server.services.base import LLMProvider
+
+log = logging.getLogger(__name__)
+
+_QUOTE_MATCH_THRESHOLD = 0.85
+_MAX_REEXTRACTION_ATTEMPTS = 2
+_EXTRACTION_MODEL = "deepseek-v4-flash:cloud"
+
+
+class Evidence(BaseModel):
+    criterion_id: str
+    quote: str
+    signals: list[str] = Field(default_factory=list)
+
+
+class ExtractionResult(BaseModel):
+    evidence: list[Evidence] = Field(default_factory=list)
+    scoring_inconclusive: bool = False
+    attempts: int = 0
+    rejected_quotes: list[str] = Field(default_factory=list)
+
+
+def _transcript_text(turns: list[dict]) -> str:
+    parts: list[str] = []
+    for t in turns:
+        role = t.get("role", "")
+        content = t.get("content", "") or t.get("text", "")
+        if content:
+            parts.append(f"{role}: {content}")
+    return "\n".join(parts)
+
+
+def _fuzzy_contains(haystack: str, quote: str) -> bool:
+    if not quote.strip():
+        return False
+    if quote in haystack:
+        return True
+    qlen = len(quote)
+    if qlen >= len(haystack):
+        return SequenceMatcher(None, quote, haystack).ratio() >= _QUOTE_MATCH_THRESHOLD
+    best = 0.0
+    window = qlen + max(20, qlen // 4)
+    step = max(1, qlen // 4)
+    i = 0
+    while i <= len(haystack) - qlen:
+        end = min(len(haystack), i + window)
+        r = SequenceMatcher(None, quote, haystack[i:end]).ratio()
+        if r > best:
+            best = r
+            if best >= _QUOTE_MATCH_THRESHOLD:
+                return True
+        i += step
+    return best >= _QUOTE_MATCH_THRESHOLD
+
+
+def _build_prompt(turns: list[dict], rubric_criteria: list[str]) -> list[dict[str, str]]:
+    transcript = _transcript_text(turns)
+    crit_block = "\n".join(f"- {c}" for c in rubric_criteria)
+    system = (
+        "You are an evidence extraction engine for a customer-service coaching rubric. "
+        "For each rubric criterion, find the single most representative verbatim quote "
+        "from the learner's utterances in the transcript, plus the observable behavior "
+        "signal tags that apply. Quotes MUST be copied verbatim from the learner's "
+        "spoken turns — do not paraphrase, do not invent."
+    )
+    user = (
+        f"Rubric criteria:\n{crit_block}\n\n"
+        f"Transcript:\n{transcript}\n\n"
+        "Return ONLY a JSON array. Each element: "
+        '{"criterion_id": <string>, "quote": <verbatim learner quote>, '
+        '"signals": [<string>, ...]}. '
+        "Omit a criterion if no evidence is present. No prose, no markdown fences."
+    )
+    return [{"role": "system", "content": system}, {"role": "user", "content": user}]
+
+
+def _parse_evidence_json(raw: str, allowed_criteria: list[str]) -> list[Evidence]:
+    text = raw.strip()
+    if text.startswith("```"):
+        text = text.strip("`")
+        if text.lower().startswith("json"):
+            text = text[4:]
+        text = text.strip()
+    try:
+        data = json.loads(text)
+    except json.JSONDecodeError as exc:
+        raise ValueError(f"evidence JSON parse failed: {exc}") from exc
+    if not isinstance(data, list):
+        raise ValueError("evidence JSON must be a list")
+    allowed = set(allowed_criteria)
+    out: list[Evidence] = []
+    for item in data:
+        try:
+            ev = Evidence.model_validate(item)
+        except ValidationError as exc:
+            raise ValueError(f"evidence item schema invalid: {exc}") from exc
+        if ev.criterion_id not in allowed:
+            raise ValueError(f"unknown criterion_id: {ev.criterion_id}")
+        out.append(ev)
+    return out
+
+
+async def extract_evidence(
+    turns: list[dict],
+    rubric_criteria: list[str],
+    llm: LLMProvider,
+    *,
+    model: str | None = None,
+    max_attempts: int = _MAX_REEXTRACTION_ATTEMPTS,
+) -> ExtractionResult:
+    """Extract verbatim-quote evidence per criterion via LLM + fuzzy verification.
+
+    Args:
+        turns: session transcript turns (each dict has role + content/text).
+        rubric_criteria: criterion ids to extract evidence for.
+        llm: LLMProvider whose chat_full returns the model's response.
+        model: override the extraction model (default deepseek-v4-flash:cloud).
+        max_attempts: max re-extraction attempts after the initial call (default 2).
+
+    Returns:
+        ExtractionResult — either with `.evidence` populated, or with
+        `.scoring_inconclusive=True` if quotes could not be verified after the
+        retry budget (grill Axis 4 MUST #3 — never silently fail to zero).
+    """
+    mdl = model or _EXTRACTION_MODEL
+    transcript_text = _transcript_text(turns)
+    rejected: list[str] = []
+    attempts = 0
+
+    for attempt in range(max_attempts + 1):
+        attempts = attempt + 1
+        messages = _build_prompt(turns, rubric_criteria)
+        if attempt > 0 and rejected:
+            messages.append(
+                {
+                    "role": "user",
+                    "content": (
+                        "The following quotes were NOT found verbatim in the transcript "
+                        "and must be replaced with exact learner utterances:\n- "
+                        + "\n- ".join(rejected[-6:])
+                        + "\n\nRe-emit the full JSON array with corrected verbatim quotes."
+                    ),
+                }
+            )
+
+        try:
+            raw, _usage = await llm.chat_full(messages, model=mdl, no_think=True)
+        except Exception as exc:
+            log.warning("evidence extraction LLM call failed (attempt %d): %s", attempts, exc)
+            continue
+
+        try:
+            candidates = _parse_evidence_json(raw, rubric_criteria)
+        except ValueError as exc:
+            log.warning("evidence JSON invalid (attempt %d): %s", attempts, exc)
+            continue
+
+        verified: list[Evidence] = []
+        bad: list[str] = []
+        for ev in candidates:
+            if _fuzzy_contains(transcript_text, ev.quote):
+                verified.append(ev)
+            else:
+                bad.append(ev.quote)
+
+        if not bad and verified:
+            return ExtractionResult(evidence=verified, attempts=attempts, rejected_quotes=rejected)
+        rejected.extend(bad)
+        if not verified and not bad:
+            continue
+
+    log.error(
+        "evidence extraction scoring_inconclusive after %d attempts; rejected=%r",
+        attempts,
+        rejected,
+    )
+    return ExtractionResult(
+        evidence=[],
+        scoring_inconclusive=True,
+        attempts=attempts,
+        rejected_quotes=rejected,
+    )
+
+
+__all__ = ["Evidence", "ExtractionResult", "extract_evidence"]

server/mastery/irt.py

@@ -0,0 +1,98 @@
+"""IRT engine — 1PL/Rasch with Bayesian theta update (SLICE-04, REQ-NFR-IRT-01).
+
+P_success(theta, b) = logistic(theta - b) = 1 / (1 + exp(-(theta - b))).
+update_theta uses a Gaussian-approximation Bayesian update (Kalman-like):
+the posterior precision is the prior precision plus the Fisher information
+P*(1-P), and the posterior mean shifts toward the outcome by the Kalman gain.
+
+Cold-start (R-IRT-01): theta=0, sigma_sq=1; until >=5 observations, scenario
+selection falls back to difficulty-based matching (difficulty closest to
+round(theta + logit(target_p))).
+"""
+
+from __future__ import annotations
+
+import math
+
+from server.scenarios.library import ScenarioLibrary
+from server.scenarios.schema import Scenario
+
+COLD_START_MIN_OBSERVATIONS = 5
+DEFAULT_THETA = 0.0
+DEFAULT_SIGMA_SQ = 1.0
+
+
+def _logit(p: float) -> float:
+    return math.log(p / (1.0 - p))
+
+
+class IRTEngine:
+    """1PL/Rasch IRT with Gaussian-approximation Bayesian theta updates."""
+
+    @staticmethod
+    def P_success(theta: float, b: float) -> float:
+        exp_neg = math.exp(-(theta - b))
+        return 1.0 / (1.0 + exp_neg)
+
+    @staticmethod
+    def update_theta(
+        theta: float, sigma_sq: float, outcome: float, b: float
+    ) -> tuple[float, float]:
+        """Bayesian update of theta given a binary (0/1) outcome.
+
+        Uses the standard 1PL Gaussian-approximation (Kalman-like) update:
+            P = P_success(theta, b)
+            new_precision = 1/sigma_sq + P*(1-P)
+            new_sigma_sq = 1 / new_precision
+            new_theta = theta + new_sigma_sq * (outcome - P)
+        """
+        p = IRTEngine.P_success(theta, b)
+        prior_precision = 1.0 / sigma_sq
+        info = p * (1.0 - p)
+        new_precision = prior_precision + info
+        new_sigma_sq = 1.0 / new_precision
+        new_theta = theta + new_sigma_sq * (outcome - p)
+        return new_theta, new_sigma_sq
+
+    @staticmethod
+    def select_scenario(
+        theta: float,
+        library: ScenarioLibrary,
+        path: str,
+        target_p: float = 0.7,
+        observations: int = 0,
+    ) -> Scenario | None:
+        """Select the next scenario for a learner.
+
+        If observations < COLD_START_MIN_OBSERVATIONS (R-IRT-01), fall back to
+        difficulty-based selection: pick the scenario whose `difficulty` is
+        closest to round(theta + logit(target_p)).
+
+        Otherwise delegate to library.select_for_theta (IRT-aware selection
+        targeting ~target_p).
+        """
+        if observations < COLD_START_MIN_OBSERVATIONS:
+            entries = library.list_by_path(path)
+            if not entries:
+                return None
+            target_difficulty = round(theta + _logit(target_p))
+            target_difficulty = max(1, min(5, target_difficulty))
+            best_entry = None
+            best_dist = math.inf
+            for e in entries:
+                dist = abs(e.difficulty - target_difficulty)
+                if dist < best_dist:
+                    best_dist = dist
+                    best_entry = e
+            if best_entry is None:
+                return None
+            return library.get(best_entry.id)
+        return library.select_for_theta(theta, path, target_p=target_p)
+
+
+__all__ = [
+    "IRTEngine",
+    "COLD_START_MIN_OBSERVATIONS",
+    "DEFAULT_THETA",
+    "DEFAULT_SIGMA_SQ",
+]

server/mastery/mastery_score.py

@@ -0,0 +1,94 @@
+"""Mastery score + gate logic — deterministic (SLICE-03 TASK-03-03).
+
+Weighted mean of per-criterion levels with a conjunctive floor (every criterion
+>= 2 AND scenario mean >= 3.0 to pass). Path score is the mean over passing
+scenarios only. Gate opens at >=3 distinct passed scenarios AND path score
+>= 3.5 (D-032).
+"""
+
+from __future__ import annotations
+
+from pydantic import BaseModel, Field
+
+from server.mastery.rubric_scorer import CriterionScore
+from server.mastery.rubric_schema import Rubric
+
+_SCENARIO_PASS_MEAN = 3.0
+_CONJUNCTIVE_FLOOR = 2
+_GATE_REQUIRED_DISTINCT = 3
+_GATE_REQUIRED_SCORE = 3.5
+
+
+class ScenarioScore(BaseModel):
+    criterion_scores: list[CriterionScore]
+    weighted_mean: float
+    passed: bool
+    fail_reason: str | None = None
+
+    @property
+    def scenario_id(self) -> str | None:
+        return None
+
+
+def compute_scenario_score(
+    criterion_scores: list[CriterionScore], rubric: Rubric
+) -> ScenarioScore:
+    """Compute a deterministic scenario score with conjunctive-floor enforcement.
+
+    Pass requires: weighted mean >= 3.0 AND every criterion >= 2 AND any
+    criterion with `conjunctive_floor` set must be >= that floor.
+    """
+    weights = {c.id: c.weight for c in rubric.criteria}
+    total = 0.0
+    for cs in criterion_scores:
+        w = weights.get(cs.criterion_id, cs.weight)
+        total += cs.level * w
+    mean = round(total, 6)
+
+    floor_violations: list[str] = []
+    for cs in criterion_scores:
+        c = rubric.criterion_by_id(cs.criterion_id)
+        floor = c.conjunctive_floor if c else None
+        required = max(floor or _CONJUNCTIVE_FLOOR, _CONJUNCTIVE_FLOOR)
+        if cs.level < required:
+            floor_violations.append(cs.criterion_id)
+
+    fail_reason: str | None = None
+    if floor_violations:
+        fail_reason = f"conjunctive_floor_violation:{','.join(floor_violations)}"
+    elif mean < _SCENARIO_PASS_MEAN:
+        fail_reason = f"mean_below_threshold:{mean}<{_SCENARIO_PASS_MEAN}"
+
+    passed = fail_reason is None
+    return ScenarioScore(
+        criterion_scores=criterion_scores,
+        weighted_mean=mean,
+        passed=passed,
+        fail_reason=fail_reason,
+    )
+
+
+def compute_path_score(passing_scenario_scores: list[ScenarioScore]) -> float:
+    """Mean weighted-mean over passing scenarios only. Empty → 0.0."""
+    if not passing_scenario_scores:
+        return 0.0
+    return round(sum(s.weighted_mean for s in passing_scenario_scores) / len(passing_scenario_scores), 6)
+
+
+def check_gate(
+    path_score: float,
+    distinct_passed_count: int,
+    *,
+    required: int = _GATE_REQUIRED_DISTINCT,
+    threshold: float = _GATE_REQUIRED_SCORE,
+) -> bool:
+    """Gate opens at >= `required` distinct passed scenarios AND path_score >= `threshold` (D-032)."""
+    return distinct_passed_count >= required and path_score >= threshold
+
+
+__all__ = [
+    "ScenarioScore",
+    "compute_scenario_score",
+    "compute_path_score",
+    "check_gate",
+]

server/mastery/rubric_loader.py

@@ -0,0 +1,65 @@
+"""Rubric loader — YAML → Pydantic Rubric (SLICE-01, D-039).
+
+Loads a competency rubric by skill name from the `rubrics/` directory, validates
+it against the Pydantic schema, and caches the parsed result in-memory for the
+lifetime of the process. Used by the scoring engine (SLICE-03) and the path
+engine (SLICE-05).
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+from threading import Lock
+from typing import Dict
+
+import yaml
+
+from server.mastery.rubric_schema import Rubric, ValidationError
+
+_DEFAULT_RUBRICS_DIR = Path(__file__).resolve().parent.parent.parent / "rubrics"
+
+_cache: Dict[str, Rubric] = {}
+_cache_lock = Lock()
+
+
+def load_rubric(skill: str, rubrics_dir: Path | None = None) -> Rubric:
+    """Load and validate a rubric by skill name.
+
+    Args:
+        skill: e.g. 'customer_service' (the YAML filename stem under rubrics/).
+        rubrics_dir: override the rubrics directory (default: repo /rubrics).
+
+    Returns:
+        A validated Rubric object. Cached in-memory per skill.
+
+    Raises:
+        FileNotFoundError: if the YAML file doesn't exist.
+        ValidationError: if the YAML fails schema validation (typed Pydantic error).
+    """
+    with _cache_lock:
+        cached = _cache.get(skill)
+        if cached is not None:
+            return cached
+
+    base = rubrics_dir or _DEFAULT_RUBRICS_DIR
+    path = base / f"{skill}.yaml"
+    if not path.exists():
+        raise FileNotFoundError(f"Rubric YAML not found: {skill} in {base}")
+
+    with path.open("r", encoding="utf-8") as f:
+        raw = yaml.safe_load(f)
+
+    rubric = Rubric.model_validate(raw)
+
+    with _cache_lock:
+        _cache[skill] = rubric
+    return rubric
+
+
+def clear_cache() -> None:
+    """Clear the in-memory rubric cache (test helper)."""
+    with _cache_lock:
+        _cache.clear()
+
+
+__all__ = ["load_rubric", "clear_cache", "ValidationError"]

server/mastery/rubric_schema.py

@@ -0,0 +1,115 @@
+"""Praxis competency rubric schema — YAML → Pydantic (SLICE-01, D-039).
+
+Defines the typed model for a competency rubric: 4+ criteria, each with 5
+behavioral anchor levels (Dreyfus + Miller "Does" + EPA entrustment per
+RESEARCH §2). Loaded from `rubrics/<skill>.yaml` by rubric_loader.py and
+referenced by the scoring engine (SLICE-03).
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+from pydantic import BaseModel, Field, ValidationError, field_validator, model_validator
+
+_LEVEL_FLOOR = 1
+_LEVEL_CEIL = 5
+_REQUIRED_LEVELS = 5
+_WEIGHT_TOLERANCE = 1e-6
+
+
+class RubricLevel(BaseModel):
+    """One anchor level (1=fail … 5=mastery/entrustable)."""
+
+    level: int = Field(..., ge=_LEVEL_FLOOR, le=_LEVEL_CEIL, description="1-5 level")
+    label: str = Field(..., description="Short human label, e.g. 'Fail', 'Mastery / Entrustable'")
+    anchor: str = Field(..., description="Observable-behavior anchor text (transcript-grounded)")
+    signals: list[str] = Field(
+        ..., min_length=1, description="Observable behavior tags that map evidence to this level"
+    )
+
+
+class RubricCriterion(BaseModel):
+    """One scoring criterion (e.g. empathy) with weight + 5 anchor levels."""
+
+    id: str = Field(..., description="Criterion id, e.g. 'empathy'")
+    name: str = Field(..., description="Human-readable criterion name")
+    weight: float = Field(..., ge=0.0, le=1.0, description="Criterion weight (sums to 1.0 across criteria)")
+    conjunctive_floor: int | None = Field(
+        None,
+        ge=_LEVEL_FLOOR,
+        le=_LEVEL_CEIL,
+        description="If set, scenario cannot pass unless this criterion ≥ floor (professionalism ≥2)",
+    )
+    levels: list[RubricLevel] = Field(..., min_length=_REQUIRED_LEVELS, max_length=_REQUIRED_LEVELS)
+
+    @field_validator("levels")
+    @classmethod
+    def _levels_are_sequential(cls, v: list[RubricLevel]) -> list[RubricLevel]:
+        seen = sorted(lvl.level for lvl in v)
+        expected = list(range(_LEVEL_FLOOR, _LEVEL_CEIL + 1))
+        if seen != expected:
+            raise ValueError(
+                f"criterion levels must be exactly 1..{_REQUIRED_LEVELS}, got {seen}"
+            )
+        return v
+
+    def level_by_value(self, level: int) -> RubricLevel | None:
+        for lvl in self.levels:
+            if lvl.level == level:
+                return lvl
+        return None
+
+
+class Rubric(BaseModel):
+    """A competency rubric for a skill (e.g. customer_service)."""
+
+    id: str = Field(..., description="Rubric id, e.g. 'customer_service'")
+    skill: str = Field(..., description="Skill path this rubric scores, e.g. 'customer_service'")
+    description: str | None = Field(None, description="Optional human description")
+    criteria: list[RubricCriterion] = Field(..., min_length=1)
+    archetype_weights: dict[str, dict[str, float]] | None = Field(
+        None, description="Per-archetype weight overrides (D-039 amendment)"
+    )
+    escalated_weights: dict[str, float] | None = Field(
+        None, description="Optional re-weight set when the escalate branch triggers (RESEARCH §6.3)"
+    )
+
+    @model_validator(mode="after")
+    def _validate_weights_and_ids(self) -> Rubric:
+        total = sum(c.weight for c in self.criteria)
+        if abs(total - 1.0) > _WEIGHT_TOLERANCE:
+            raise ValueError(
+                f"criterion weights must sum to 1.0 (±{_WEIGHT_TOLERANCE}), got {total}"
+            )
+        ids = [c.id for c in self.criteria]
+        if len(ids) != len(set(ids)):
+            dupes = sorted({i for i in ids if ids.count(i) > 1})
+            raise ValueError(f"duplicate criterion ids: {dupes}")
+        if self.skill != self.id and not self.id.startswith(self.skill):
+            pass
+        return self
+
+    def criterion_by_id(self, criterion_id: str) -> RubricCriterion | None:
+        for c in self.criteria:
+            if c.id == criterion_id:
+                return c
+        return None
+
+    def weights_for_archetype(self, archetype: str | None) -> dict[str, float]:
+        """Return {criterion_id: weight} for an archetype, falling back to the base weights."""
+        if archetype and self.archetype_weights and archetype in self.archetype_weights:
+            override = self.archetype_weights[archetype]
+            return {c.id: override.get(c.id, c.weight) for c in self.criteria}
+        return {c.id: c.weight for c in self.criteria}
+
+    def criterion_ids(self) -> list[str]:
+        return [c.id for c in self.criteria]
+
+
+__all__ = [
+    "Rubric",
+    "RubricCriterion",
+    "RubricLevel",
+    "ValidationError",
+]

server/mastery/rubric_scorer.py

@@ -0,0 +1,67 @@
+"""Rule-based rubric scorer — deterministic (SLICE-03 TASK-03-02, REQ-NFR-MAST-01).
+
+No LLM. Maps evidence signals to rubric level anchors: for each criterion, pick
+the highest level whose `signals[]` are all present in the matched evidence,
+fallback to level 1 if no level matches. The output is reproducible given the
+same (evidence, rubric) pair.
+"""
+
+from __future__ import annotations
+
+from pydantic import BaseModel, Field
+
+from server.mastery.evidence_extractor import Evidence
+from server.mastery.rubric_schema import Rubric, RubricCriterion
+
+
+class CriterionScore(BaseModel):
+    criterion_id: str
+    level: int = Field(ge=1, le=5)
+    weight: float
+    evidence_quote: str = ""
+    matched_signals: list[str] = Field(default_factory=list)
+
+
+def _evidence_for(evidence: list[Evidence], criterion_id: str) -> Evidence | None:
+    for ev in evidence:
+        if ev.criterion_id == criterion_id:
+            return ev
+    return None
+
+
+def _level_for_criterion(criterion: RubricCriterion, ev: Evidence | None) -> tuple[int, list[str]]:
+    if ev is None or not ev.signals:
+        return 1, []
+    ev_signals = set(ev.signals)
+    best_level = 1
+    best_signals: list[str] = []
+    for lvl in sorted(criterion.levels, key=lambda l: l.level):
+        if all(s in ev_signals for s in lvl.signals):
+            best_level = lvl.level
+            best_signals = list(lvl.signals)
+    return best_level, best_signals
+
+
+def score(evidence: list[Evidence], rubric: Rubric) -> list[CriterionScore]:
+    """Score evidence against the rubric — deterministic, no LLM.
+
+    Returns one CriterionScore per rubric criterion, in rubric order. Criteria
+    with no matching evidence get level 1 (the "Fail" anchor).
+    """
+    out: list[CriterionScore] = []
+    for c in rubric.criteria:
+        ev = _evidence_for(evidence, c.id)
+        level, matched = _level_for_criterion(c, ev)
+        out.append(
+            CriterionScore(
+                criterion_id=c.id,
+                level=level,
+                weight=c.weight,
+                evidence_quote=ev.quote if ev else "",
+                matched_signals=matched,
+            )
+        )
+    return out
+
+
+__all__ = ["CriterionScore", "score"]

server/paths/__init__.py

@@ -0,0 +1,18 @@
+"""Praxis path engine package (SLICE-05, REQ-PATH-02).
+
+Defines the 6-week competency path structure with mastery gates (D-037),
+loaded from YAML into typed Pydantic models and driven by the path engine.
+"""
+
+from __future__ import annotations
+
+from server.paths.schema import Path, PathWeek, WeekGate, ValidationError
+from server.paths.engine import PathEngine
+
+__all__ = [
+    "Path",
+    "PathWeek",
+    "WeekGate",
+    "PathEngine",
+    "ValidationError",
+]

server/paths/engine.py

@@ -0,0 +1,158 @@
+"""Praxis path engine — 6-week progression + mastery gates (SLICE-05, REQ-PATH-02).
+
+Loads a competency path YAML, reads learner progress, checks week gates, and
+advances the learner week-by-week per D-048. Gate evaluation delegates to
+`server.mastery.mastery_score.check_gate` when available (SLICE-03); until
+then, a local deterministic gate check implements the same D-032 contract
+(>= required_scenarios distinct passed AND >= required_score mean).
+
+The loader does NOT fail when referenced scenario YAMLs are missing — the
+scenarios are authored in SLICE-06. Use `validate_scenarios_exist(library)`
+once the library is populated to enforce referential integrity.
+"""
+
+from __future__ import annotations
+
+from copy import deepcopy
+from pathlib import Path as FsPath
+from threading import Lock
+from typing import Any, Dict
+
+import yaml
+
+from server.paths.schema import Path, PathWeek, ValidationError
+
+_DEFAULT_PATHS_DIR = FsPath(__file__).resolve().parent.parent.parent / "paths"
+_MAX_WEEK = 6
+
+_cache: Dict[str, Path] = {}
+_cache_lock = Lock()
+
+
+def _local_check_gate(distinct_passed: int, mean_score: float, gate: Any) -> bool:
+    return distinct_passed >= gate.required_scenarios and mean_score >= gate.required_score
+
+
+def _resolve_mastery_check_gate():
+    try:
+        from server.mastery.mastery_score import check_gate as _ms_check_gate  # type: ignore[import]
+    except Exception:
+        return None
+    return _ms_check_gate
+
+
+def _eval_gate(progress: dict, week: int, path: Path, gate: Any) -> bool:
+    distinct_passed = int(progress.get("distinct_passed", 0))
+    mean_score = float(progress.get("mastery_score", 0.0))
+    ms_check_gate = _resolve_mastery_check_gate()
+    if ms_check_gate is not None:
+        try:
+            return bool(ms_check_gate(mean_score, distinct_passed, gate))
+        except TypeError:
+            try:
+                return bool(ms_check_gate(path_score=mean_score, distinct_passed_count=distinct_passed, gate=gate))
+            except TypeError:
+                pass
+    return _local_check_gate(distinct_passed, mean_score, gate)
+
+
+class PathEngine:
+    """Loads paths and drives 6-week progression + mastery gate evaluation."""
+
+    def __init__(self, paths_dir: FsPath | None = None) -> None:
+        self.paths_dir = paths_dir or _DEFAULT_PATHS_DIR
+
+    def load_path(self, slug: str) -> Path:
+        """Load and validate a path by slug. Cached in-memory per slug.
+
+        Does NOT validate that referenced scenarios exist (SLICE-06 authors
+        them); call `validate_scenarios_exist(library)` for that.
+        """
+        with _cache_lock:
+            cached = _cache.get(slug)
+            if cached is not None:
+                return cached
+
+        path = self.paths_dir / f"{slug}.yaml"
+        if not path.exists():
+            raise FileNotFoundError(f"Path YAML not found: {slug} in {self.paths_dir}")
+
+        with path.open("r", encoding="utf-8") as f:
+            raw = yaml.safe_load(f)
+
+        parsed = Path.model_validate(raw)
+
+        with _cache_lock:
+            _cache[slug] = parsed
+        return parsed
+
+    def validate_scenarios_exist(self, path: Path, library: Any) -> list[str]:
+        """Verify every scenario_id referenced by the path exists in the library.
+
+        Returns the list of all referenced scenario ids on success. Raises
+        ValueError listing the missing ids. Call only after SLICE-06 has
+        authored the scenarios.
+        """
+        referenced = path.all_scenario_ids()
+        missing: list[str] = []
+        for sid in referenced:
+            try:
+                library.get(sid)
+            except Exception:
+                missing.append(sid)
+        if missing:
+            raise ValueError(
+                f"path {path.slug!r} references {len(missing)} missing scenario(s): {missing}"
+            )
+        return referenced
+
+    def current_week(self, progress: dict) -> int:
+        """Read the learner's current week from mastery_progress.current_week.
+
+        Defaults to 1 (cold start) when absent or out of range.
+        """
+        w = int(progress.get("current_week", 1))
+        if w < 1:
+            return 1
+        if w > _MAX_WEEK:
+            return _MAX_WEEK
+        return w
+
+    def check_gate(self, progress: dict, week: int, path: Path) -> bool:
+        """Evaluate whether the mastery gate for `week` is open.
+
+        Reads `distinct_passed` and `mastery_score` from `progress` and
+        compares against the week's gate config (D-032). Delegates to
+        `mastery_score.check_gate` when the SLICE-03 module is importable.
+        """
+        week_obj = path.week_by_number(week)
+        if week_obj is None:
+            raise ValueError(f"week {week} not in path {path.slug!r} (weeks 1..{_MAX_WEEK})")
+        return _eval_gate(progress, week, path, week_obj.gate)
+
+    def advance_week(self, progress: dict) -> dict:
+        """Increment current_week (D-048). Returns a new progress dict.
+
+        Does NOT mutate the input. Caps at week 6. The caller is expected to
+        have verified the current week's gate is open before calling.
+        """
+        out = deepcopy(progress)
+        w = self.current_week(out)
+        if w < _MAX_WEEK:
+            out["current_week"] = w + 1
+        else:
+            out["current_week"] = _MAX_WEEK
+        return out
+
+    def is_path_complete(self, progress: dict, path: Path) -> bool:
+        """True when the week-6 mastery gate is open (path fully complete)."""
+        return self.check_gate(progress, _MAX_WEEK, path)
+
+
+def clear_cache() -> None:
+    """Clear the in-memory path cache (test helper)."""
+    with _cache_lock:
+        _cache.clear()
+
+
+__all__ = ["PathEngine", "Path", "PathWeek", "ValidationError", "clear_cache"]

server/paths/schema.py

@@ -0,0 +1,105 @@
+"""Praxis path schema — YAML DSL -> Pydantic (SLICE-05, D-037, REQ-PATH-02).
+
+Defines the typed model for a 6-week competency path. Each week lists the
+scenarios it exercises and a mastery gate (>= required_scenarios distinct
+scenarios passed, >= required_score mean score per D-032). Loaded from
+`paths/<slug>.yaml` by server/paths/engine.py.
+
+Per PRD section 6.4 (D-037): exactly 6 weeks, numbered 1..6 sequentially.
+"""
+
+from __future__ import annotations
+
+from pydantic import BaseModel, Field, ValidationError, field_validator, model_validator
+
+_REQUIRED_WEEKS = 6
+_MIN_WEEK = 1
+_MAX_WEEK = 6
+_DEFAULT_REQUIRED_SCENARIOS = 3
+_DEFAULT_REQUIRED_SCORE = 3.5
+
+
+class WeekGate(BaseModel):
+    """Mastery gate config for one week (D-032).
+
+    A week's gate opens when the learner has passed >= required_scenarios
+    distinct scenarios with a mean score >= required_score across those
+    passing scenarios.
+    """
+
+    required_scenarios: int = Field(
+        _DEFAULT_REQUIRED_SCENARIOS,
+        ge=1,
+        description="Min distinct passed scenarios to open the gate (D-032 default 3)",
+    )
+    required_score: float = Field(
+        _DEFAULT_REQUIRED_SCORE,
+        ge=0.0,
+        description="Min mean score across passing scenarios to open the gate (D-032 default 3.5)",
+    )
+
+
+class PathWeek(BaseModel):
+    """One week in a 6-week competency path."""
+
+    week: int = Field(..., ge=_MIN_WEEK, le=_MAX_WEEK, description="Week number 1..6")
+    title: str = Field(..., min_length=1, description="Human-readable week title")
+    scenario_ids: list[str] = Field(
+        ..., min_length=1, description="Scenario ids exercised this week (authored in SLICE-06)"
+    )
+    gate: WeekGate = Field(default_factory=WeekGate, description="Mastery gate for this week")
+
+    @field_validator("scenario_ids")
+    @classmethod
+    def _scenario_ids_unique(cls, v: list[str]) -> list[str]:
+        if len(v) != len(set(v)):
+            dupes = sorted({s for s in v if v.count(s) > 1})
+            raise ValueError(f"duplicate scenario_ids in week: {dupes}")
+        return v
+
+
+class Path(BaseModel):
+    """A 6-week competency path (D-037, PRD section 6.4)."""
+
+    slug: str = Field(..., min_length=1, description="Path slug, e.g. 'customer_service'")
+    name: str = Field(..., min_length=1, description="Human-readable path name")
+    skill: str = Field(..., min_length=1, description="Skill this path develops (matches a rubric id)")
+    weeks: list[PathWeek] = Field(..., description="Exactly 6 weeks, numbered 1..6 sequentially")
+
+    @model_validator(mode="after")
+    def _validate_weeks(self) -> Path:
+        if len(self.weeks) != _REQUIRED_WEEKS:
+            raise ValueError(
+                f"path must have exactly {_REQUIRED_WEEKS} weeks (D-037 / PRD section 6.4), "
+                f"got {len(self.weeks)}"
+            )
+        seen = sorted(w.week for w in self.weeks)
+        expected = list(range(_MIN_WEEK, _MAX_WEEK + 1))
+        if seen != expected:
+            raise ValueError(
+                f"week numbers must be exactly 1..{_REQUIRED_WEEKS} sequential, got {seen}"
+            )
+        dupes = [w.week for w in self.weeks if [x.week for x in self.weeks].count(w.week) > 1]
+        if dupes:
+            raise ValueError(f"duplicate week numbers: {sorted(set(dupes))}")
+        return self
+
+    def week_by_number(self, week: int) -> PathWeek | None:
+        for w in self.weeks:
+            if w.week == week:
+                return w
+        return None
+
+    def all_scenario_ids(self) -> list[str]:
+        ids: list[str] = []
+        for w in self.weeks:
+            ids.extend(w.scenario_ids)
+        return ids
+
+
+__all__ = [
+    "Path",
+    "PathWeek",
+    "WeekGate",
+    "ValidationError",
+]

server/scenarios/library.py

@@ -0,0 +1,194 @@
+"""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",
+]

server/scenarios/loader.py

@@ -16,11 +16,42 @@ from server.scenarios.schema import Scenario, ValidationError
 _DEFAULT_SCENARIOS_DIR = Path(__file__).resolve().parent.parent.parent / "scenarios"
 
 
+def _find_yaml(scenario_id: str, base: Path) -> Path | None:
+    """Resolve a scenario id to its YAML path.
+
+    Searches the scenarios root and any one-level subdirectory (e.g.
+    customer_service/). Supports two alias forms for backward compatibility:
+      - cs_<id> -> customer_service_<id>.yaml  (v0.1 call sites used the long form)
+      - customer_service_<id> -> cs_<id>.yaml  (reverse, for the renamed v01 file)
+    """
+    primary = base / f"{scenario_id}.yaml"
+    if primary.exists():
+        return primary
+    cs_alias = base / f"{scenario_id.replace('cs_', 'customer_service_')}.yaml"
+    if cs_alias.exists():
+        return cs_alias
+    long_alias = base / f"{scenario_id.replace('customer_service_', 'cs_')}.yaml"
+    if long_alias.exists():
+        return long_alias
+    # One-level subdirectory walk (subdir named by skill, e.g. customer_service/).
+    for d in sorted(base.glob("*/")):
+        if not d.is_dir():
+            continue
+        for cand in (
+            d / f"{scenario_id}.yaml",
+            d / f"{scenario_id.replace('cs_', 'customer_service_')}.yaml",
+            d / f"{scenario_id.replace('customer_service_', 'cs_')}.yaml",
+        ):
+            if cand.exists():
+                return cand
+    return None
+
+
 def load(scenario_id: str, scenarios_dir: Path | None = None) -> Scenario:
     """Load and validate a scenario by id.
 
     Args:
-        scenario_id: e.g. 'customer_service_refund_ca_v01' (the YAML filename stem).
+        scenario_id: e.g. 'cs_refund_ca_v01' (the YAML filename stem).
         scenarios_dir: override the scenarios directory (default: repo /scenarios).
 
     Returns:
@@ -31,12 +62,9 @@ def load(scenario_id: str, scenarios_dir: Path | None = None) -> Scenario:
         ValidationError: if the YAML fails schema validation (typed Pydantic error).
     """
     base = scenarios_dir or _DEFAULT_SCENARIOS_DIR
-    path = base / f"{scenario_id}.yaml"
-    if not path.exists():
-        # Try the id-with-cs-prefix alias (RESEARCH example used 'cs_refund_ca_v01').
-        path = base / f"{scenario_id.replace('cs_', 'customer_service_')}.yaml"
-        if not path.exists():
-            raise FileNotFoundError(f"Scenario YAML not found: {scenario_id} in {base}")
+    path = _find_yaml(scenario_id, base)
+    if path is None:
+        raise FileNotFoundError(f"Scenario YAML not found: {scenario_id} in {base}")
 
     with path.open("r", encoding="utf-8") as f:
         raw = yaml.safe_load(f)
@@ -45,10 +73,15 @@ def load(scenario_id: str, scenarios_dir: Path | None = None) -> Scenario:
 
 
 def load_all(scenarios_dir: Path | None = None) -> list[Scenario]:
-    """Load all scenarios in the directory (for the future scenario library)."""
+    """Load all scenarios in the directory tree (root + one-level subdirs)."""
     base = scenarios_dir or _DEFAULT_SCENARIOS_DIR
     out: list[Scenario] = []
-    for p in sorted(base.glob("*.yaml")):
+    paths = sorted(base.glob("*.yaml")) + sorted(base.glob("*/**/*.yaml"))
+    seen: set[Path] = set()
+    for p in paths:
+        if p in seen or p.name == "index.yaml" or p.name == "cost_rates.yaml":
+            continue
+        seen.add(p)
         with p.open("r", encoding="utf-8") as f:
             raw = yaml.safe_load(f)
         out.append(Scenario.model_validate(raw))

server/scenarios/schema.py

@@ -9,9 +9,12 @@ accept), failure_mode field present (D-009 — not provoked in v0.1).
 
 from __future__ import annotations
 
+import re
 from typing import Literal
 
-from pydantic import BaseModel, Field, ValidationError
+from pydantic import BaseModel, Field, ValidationError, field_validator
+
+_SEMVER_RE = re.compile(r"^\d+\.\d+\.\d+(?:-[0-9A-Za-z.-]+)?(?:\+[0-9A-Za-z.-]+)?$")
 
 
 class ScenarioPersona(BaseModel):
@@ -60,8 +63,28 @@ class ScenarioDebrief(BaseModel):
     )
 
 
+class RubricMapping(BaseModel):
+    """Maps a scenario to one rubric criterion (SLICE-02 — D-039).
+
+    A scenario lists the rubric criteria it exercises; the scoring engine
+    (SLICE-03) extracts evidence for each and scores against the rubric YAML.
+    """
+
+    criterion_id: str = Field(..., description="Rubric criterion id, e.g. 'empathy'")
+    weight: float | None = Field(
+        None, description="Optional per-scenario weight override (defaults to rubric weight)"
+    )
+    evidence_required: bool = Field(
+        True, description="If True, the scorer must find evidence to score this criterion"
+    )
+
+
 class Scenario(BaseModel):
-    """A Praxis role-play scenario (D-018 — YAML → Pydantic → Pipecat Flows)."""
+    """A Praxis role-play scenario (D-018 — YAML → Pydantic → Pipecat Flows).
+
+    Extended in v0.3 (SLICE-02) with rubric mapping + IRT + provenance fields.
+    All new fields have defaults so v0.1 scenario YAMLs still load unchanged.
+    """
 
     id: str = Field(..., description="Scenario id, e.g. 'cs_refund_ca_v01'")
     path: str = Field(..., description="Skill path, e.g. 'customer_service'")
@@ -79,6 +102,28 @@ class Scenario(BaseModel):
     branches: list[Branch] = Field(..., min_length=1, description="Branch points (v0.1: 2)")
     debrief: ScenarioDebrief
 
+    rubric_criteria: list[RubricMapping] = Field(
+        default_factory=list,
+        description="Rubric criteria this scenario exercises (SLICE-02). Empty for v0.1 scenarios.",
+    )
+    irt_target_p: float = Field(
+        0.7, ge=0.0, le=1.0, description="Target P for IRT scenario selection (D-035 default 0.7)"
+    )
+    version: str = Field("1.0.0", description="Scenario semver (D-036)")
+    generated_from: str | None = Field(
+        None, description="AI-variation backref: parent scenario id if this was generated (D-036)"
+    )
+    intent_hash: str | None = Field(
+        None, description="Structural drift detection hash (D-036)"
+    )
+
+    @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
+
     def branch_ids(self) -> list[str]:
         return [b.id for b in self.branches]
 
@@ -88,6 +133,9 @@ class Scenario(BaseModel):
                 return b
         return None
 
+    def rubric_criterion_ids(self) -> list[str]:
+        return [m.criterion_id for m in self.rubric_criteria]
+
 
 __all__ = [
     "Scenario",
@@ -96,5 +144,6 @@ __all__ = [
     "Branch",
     "BranchTrigger",
     "ScenarioDebrief",
+    "RubricMapping",
     "ValidationError",
 ]

server/session_recorder.py

@@ -5,16 +5,27 @@ Per turn: log a turns row with ASR/TTS text + latency.
 On branch decision: update branch_path.
 On session end: set outcome + update progress + store cost + debrief.
 
+After end(): the caller may invoke `run_mastery_flow()` to run the off-voice-path
+mastery scoring pipeline (SLICE-07 TASK-07-01): evidence extraction → rubric
+scoring → scenario score → IRT theta update → path gate check + week advance →
+SQLite gate-event audit → optional VC issuance (SLICE-09, lazy import).
+
 No auth — learner_id is the hardcoded 'learner-1' (D-007).
 """
 
 from __future__ import annotations
 
-from typing import Any
+import asyncio
+import json
+import logging
+import uuid
+from typing import Any, Awaitable, Callable
 
 from db.store import PraxisStore, HARDCODED_LEARNER_ID
 from server.cost import CostBreakdown, derive_cost
 
+log = logging.getLogger(__name__)
+
 
 class SessionRecorder:
     """Records a voice session to SQLite (TASK-04-03)."""
@@ -38,6 +49,11 @@ class SessionRecorder:
         self._debrief_input_tokens = 0
         self._debrief_output_tokens = 0
         self._branch_path: list[str] = []
+        # Transcribed turns captured for the post-session mastery flow.
+        # Each entry: {"role": "learner"|"customer"|"assistant", "content": str}.
+        self._mastery_turns: list[dict[str, str]] = []
+        # Populated by run_mastery_flow(); surfaced to the debrief caller.
+        self.mastery_result: dict[str, Any] | None = None
 
     async def start(self) -> str:
         """Create the session row; return the session id."""
@@ -62,9 +78,12 @@ class SessionRecorder:
         if asr_text:
             # Rough: 1 token ≈ 4 chars.
             self._llm_input_tokens += len(asr_text) // 4
+            self._mastery_turns.append({"role": role, "content": asr_text})
         if tts_text:
             self._tts_chars += len(tts_text)
             self._llm_output_tokens += len(tts_text) // 4
+            if role == "assistant" and not asr_text:
+                self._mastery_turns.append({"role": role, "content": tts_text})
         if latency_ms and role == "assistant":
             # Rough audio-minutes estimate from latency (placeholder for real metering).
             pass
@@ -79,13 +98,24 @@ class SessionRecorder:
     def set_branch_path(self, branch_path: list[str]) -> None:
         self._branch_path = branch_path
 
+    def set_mastery_turns(self, turns: list[dict[str, str]]) -> None:
+        """Override the captured transcript turns used by run_mastery_flow()."""
+        self._mastery_turns = list(turns)
+
     async def end(
         self,
         outcome: str,
         tts_provider: str = "cartesia",
         debrief_text: str | None = None,
+        schedule_mastery: bool = False,
+        mastery_deps: "MasteryFlowDeps | None" = None,
     ) -> CostBreakdown:
-        """End the session: derive cost, write the session row, update progress."""
+        """End the session: derive cost, write the session row, update progress.
+
+        If `schedule_mastery=True` and `mastery_deps` is provided, the mastery
+        flow is scheduled as a fire-and-forget asyncio task (off the voice
+        path). The task result lands in `self.mastery_result` once it completes.
+        """
         if self.session_id is None:
             raise RuntimeError("SessionRecorder.end() called before start()")
 
@@ -108,7 +138,201 @@ class SessionRecorder:
             debrief_text=debrief_text,
         )
         await self.store.update_progress(self.learner_id, self.scenario_id, outcome)
+
+        if schedule_mastery and mastery_deps is not None:
+            asyncio.create_task(
+                self._run_mastery_flow_guarded(mastery_deps)
+            )
         return breakdown
 
+    async def _run_mastery_flow_guarded(self, deps: "MasteryFlowDeps") -> None:
+        try:
+            await self.run_mastery_flow(deps)
+        except Exception:
+            log.exception("mastery flow failed for session %s", self.session_id)
 
-__all__ = ["SessionRecorder"]
+    async def run_mastery_flow(self, deps: "MasteryFlowDeps") -> dict[str, Any]:
+        """Run the off-voice-path mastery scoring pipeline (SLICE-07 TASK-07-01).
+
+        Steps:
+          1. evidence_extractor.extract_evidence(turns, rubric_criteria, llm)
+          2. if ExtractionResult.scoring_inconclusive → return inconclusive
+             status (no score, no gate event, no progress change). The caller
+             surfaces a retry in the debrief (grill Axis 4 MUST #3).
+          3. rubric_scorer.score(evidence, rubric)
+          4. mastery_score.compute_scenario_score(criterion_scores, rubric)
+          5. irt.update_theta + persist via store.upsert_ability
+          6. path_engine.check_gate + advance_week + persist via store.upsert_progress
+          7. record mastery_gate_event in SQLite (audit, REQ-NFR-MAST-02)
+          8. if week-final gate open → vc_issuer.issue_credential (lazy import;
+             SLICE-09 may not be present yet → ImportError is swallowed)
+
+        Returns a dict describing the result (status, scenario_score, theta,
+        week, gate_open, ...). Stored on `self.mastery_result`.
+        """
+        from server.mastery import evidence_extractor as _ev
+        from server.mastery import mastery_score as _ms
+        from server.mastery import rubric_scorer as _rs
+
+        rubric = deps.load_rubric()
+        scenario = deps.load_scenario()
+        criterion_ids = [m.criterion_id for m in scenario.rubric_criteria] or rubric.criterion_ids()
+        path_slug = scenario.path
+
+        extraction = await _ev.extract_evidence(
+            self._mastery_turns, criterion_ids, deps.llm
+        )
+        if extraction.scoring_inconclusive:
+            self.mastery_result = {
+                "status": "scoring_inconclusive",
+                "attempts": extraction.attempts,
+                "rejected_quotes": extraction.rejected_quotes,
+                "retry_advised": True,
+            }
+            return self.mastery_result
+
+        criterion_scores = _rs.score(extraction.evidence, rubric)
+        scenario_score = _ms.compute_scenario_score(criterion_scores, rubric)
+
+        progress_row = await self.store.get_progress(self.learner_id, path_slug)
+        if progress_row is not None:
+            progress = dict(progress_row)
+            scenarios_passed: list[str] = list(
+                json.loads(progress.get("scenarios_passed_json") or "[]")
+            )
+        else:
+            progress = {}
+            scenarios_passed = []
+        if scenario_score.passed and self.scenario_id not in scenarios_passed:
+            scenarios_passed.append(self.scenario_id)
+        # Recompute the path score over the passing set we know about.
+        path_score = _ms.compute_path_score(
+            [scenario_score] if scenario_score.passed else []
+        )
+        # If prior passing scenario scores are tracked elsewhere, they'd be
+        # folded in here; the mastery_progress row stores the cumulative mean.
+
+        path = deps.load_path()
+        week = deps.path_engine.current_week(progress) if progress else 1
+        gate_open = deps.path_engine.check_gate(
+            {"distinct_passed": len(scenarios_passed), "mastery_score": path_score},
+            week,
+            path,
+        )
+
+        # IRT theta update (uses scenario difficulty as the item parameter b).
+        ability_row = await self.store.get_ability(self.learner_id, path_slug)
+        if ability_row is not None:
+            theta = float(ability_row["theta"])
+            sigma_sq = float(ability_row["sigma_sq"])
+            observations = int(ability_row["observations"])
+        else:
+            theta = 0.0
+            sigma_sq = 1.0
+            observations = 0
+        outcome = 1.0 if scenario_score.passed else 0.0
+        b = float(scenario.difficulty)
+        new_theta, new_sigma_sq = deps.irt.update_theta(theta, sigma_sq, outcome, b)
+        new_observations = observations + 1
+        await self.store.upsert_ability(
+            self.learner_id, path_slug, new_theta, new_sigma_sq, new_observations
+        )
+
+        # Advance the week only if the gate is open (D-048).
+        new_progress = progress
+        if gate_open:
+            new_progress = deps.path_engine.advance_week(progress or {"current_week": week})
+            new_progress["distinct_passed"] = len(scenarios_passed)
+            new_progress["mastery_score"] = path_score
+        else:
+            new_progress = dict(progress or {"current_week": week})
+            new_progress["distinct_passed"] = len(scenarios_passed)
+            new_progress["mastery_score"] = path_score
+        new_week = int(new_progress.get("current_week", week))
+        await self.store.upsert_progress(
+            self.learner_id,
+            path_slug,
+            new_week,
+            scenarios_passed,
+            path_score,
+            gate_open,
+        )
+
+        # Audit log (REQ-NFR-MAST-02). scoring_inconclusive never reaches here.
+        rubric_scores_json = [cs.model_dump() for cs in criterion_scores]
+        await self.store.record_gate_event(
+            self.learner_id,
+            path_slug,
+            week,
+            scenarios_passed,
+            rubric_scores_json,
+            path_score,
+            gate_open,
+        )
+
+        # VC issuance — week-final gate open (grill Axis 8 MUST). SLICE-09 may
+        # not exist yet; the lazy import is wrapped so P1 ships independently.
+        vc_credential_id: str | None = None
+        path_complete = gate_open and new_week >= 6
+        if path_complete:
+            try:
+                from server.vc.issuer import issue_credential as _issue_credential  # type: ignore
+
+                vc_credential_id = await _issue_credential(
+                    store=self.store,
+                    learner_id=self.learner_id,
+                    path=path_slug,
+                    scenarios_passed=scenarios_passed,
+                    rubric_score=path_score,
+                    completed_weeks=new_week,
+                    evidence=rubric_scores_json,
+                )
+            except ImportError:
+                log.info("vc_issuer not available (SLICE-09 pending); skipping issuance")
+            except Exception:
+                log.exception("vc issuance failed for learner %s", self.learner_id)
+
+        self.mastery_result = {
+            "status": "scored",
+            "scenario_id": self.scenario_id,
+            "weighted_mean": scenario_score.weighted_mean,
+            "passed": scenario_score.passed,
+            "fail_reason": scenario_score.fail_reason,
+            "theta": new_theta,
+            "sigma_sq": new_sigma_sq,
+            "observations": new_observations,
+            "week": week,
+            "new_week": new_week,
+            "gate_open": gate_open,
+            "path_complete": path_complete,
+            "vc_credential_id": vc_credential_id,
+            "attempts": extraction.attempts,
+        }
+        return self.mastery_result
+
+
+class MasteryFlowDeps:
+    """Dependency bundle for SessionRecorder.run_mastery_flow().
+
+    Injected by the caller (DI): keeps session_recorder.py decoupled from the
+    concrete rubric/scenario/path loaders and the LLM provider.
+    """
+
+    def __init__(
+        self,
+        llm: Any,
+        irt: Any,
+        path_engine: Any,
+        load_rubric: Callable[[], Any],
+        load_scenario: Callable[[], Any],
+        load_path: Callable[[], Any],
+    ) -> None:
+        self.llm = llm
+        self.irt = irt
+        self.path_engine = path_engine
+        self.load_rubric = load_rubric
+        self.load_scenario = load_scenario
+        self.load_path = load_path
+
+
+__all__ = ["SessionRecorder", "MasteryFlowDeps"]

server/vc/issuer.py

@@ -0,0 +1,214 @@
+"""W3C VC 2.0 issuance — Ed25519 + JCS + eddsa-jcs-2022 proof (SLICE-09 TASK-09-02).
+
+Builds a Verifiable Credential per VC-DM 2.0, secures it with a Data Integrity
+`eddsa-jcs-2022` proof (JCS canonicalization, Ed25519 signature), and persists
+it to SQLite. The `issue_credential` coroutine is the entry point wired into
+SessionRecorder.run_mastery_flow (grill Axis 8 MUST).
+
+Credential tier is `formative` (grill Axis 4 MUST #1) — the v0.3 credential is
+a formative mastery signal, not a high-stakes summative credential.
+"""
+
+from __future__ import annotations
+
+import base64
+import datetime as _dt
+import hashlib
+import json
+import os
+import uuid
+from typing import Any
+
+import canonicaljson
+import nacl.signing
+from db.store import PraxisStore
+
+from server.vc.issuer_keys import KeyPair, get_active_signing_key
+from server.vc.status_list import BitstringStatusList
+
+ISSUER_URL_DEFAULT = "https://praxis.example/issuers/v0.3"
+CONTEXTS = [
+    "https://www.w3.org/ns/credentials/v2",
+    "https://praxis.example/contexts/mastery/v1",
+]
+CREDENTIAL_TIER = "formative"
+
+
+def _issuer_url() -> str:
+    return os.environ.get("PRAXIS_ISSUER_URL", ISSUER_URL_DEFAULT).rstrip("/")
+
+
+def _now_iso() -> str:
+    return _dt.datetime.now(_dt.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
+
+
+def _valid_until(issuance_iso: str, years: int = 3) -> str:
+    dt = _dt.datetime.strptime(issuance_iso, "%Y-%m-%dT%H:%M:%SZ").replace(
+        tzinfo=_dt.timezone.utc
+    )
+    return (dt + _dt.timedelta(days=365 * years)).strftime("%Y-%m-%dT%H:%M:%SZ")
+
+
+def build_vc_payload(
+    learner_ref: str,
+    path: str,
+    scenarios_passed: list[str],
+    rubric_score: float,
+    completed_weeks: int,
+    evidence: list[dict[str, Any]] | None,
+    credential_id: str | None = None,
+    status_list_index: int | None = None,
+) -> dict[str, Any]:
+    issuance = _now_iso()
+    issuer = _issuer_url()
+    cid = credential_id or f"vc-{uuid.uuid4().hex[:16]}"
+    payload: dict[str, Any] = {
+        "@context": list(CONTEXTS),
+        "id": f"{issuer}/vc/{cid}",
+        "type": ["VerifiableCredential", "MasteryCredential"],
+        "issuer": issuer,
+        "validFrom": issuance,
+        "validUntil": _valid_until(issuance, 3),
+        "name": f"Mastery of {path.replace('-', ' ').title()}",
+        "description": (
+            "Praxis v0.3 formative mastery credential — the holder demonstrated "
+            "competency across varied scenarios, scored against a 5-level rubric."
+        ),
+        "credentialTier": CREDENTIAL_TIER,
+        "credentialSubject": {
+            "id": f"urn:uuid:{learner_ref}",
+            "type": "Person",
+            "skill": path,
+            "level": "mastery",
+            "path": path,
+            "completedWeeks": completed_weeks,
+            "rubricScore": round(float(rubric_score), 3),
+            "rubricMax": 5.0,
+            "rubricThreshold": 3.5,
+            "scenariosPassed": list(scenarios_passed),
+            "credentialTier": CREDENTIAL_TIER,
+            "evidence": evidence or [],
+        },
+    }
+    if status_list_index is not None:
+        payload["credentialStatus"] = {
+            "type": "BitstringStatusListEntry",
+            "statusPurpose": "revocation",
+            "statusListIndex": str(status_list_index),
+            "statusListCredential": f"{issuer}/status/default",
+        }
+    return payload
+
+
+def canonicalize(payload: dict[str, Any]) -> bytes:
+    return canonicaljson.encode_canonical_json(payload)
+
+
+def _build_proof_config(key_id: str) -> dict[str, Any]:
+    issuer = _issuer_url()
+    return {
+        "type": "DataIntegrityProof",
+        "cryptosuite": "eddsa-jcs-2022",
+        "created": _now_iso(),
+        "verificationMethod": f"{issuer}/keys/{key_id}",
+        "proofPurpose": "assertionMethod",
+    }
+
+
+def _compute_hash_data(
+    unsecured_doc: dict[str, Any], proof_options: dict[str, Any]
+) -> bytes:
+    canonical_doc = canonicalize(unsecured_doc)
+    canonical_proof = canonicalize(proof_options)
+    return hashlib.sha256(canonical_proof).digest() + hashlib.sha256(
+        canonical_doc
+    ).digest()
+
+
+def sign(payload: dict[str, Any], signing_key: nacl.signing.SigningKey, key_id: str) -> tuple[dict[str, Any], str]:
+    proof_options = _build_proof_config(key_id)
+    hash_data = _compute_hash_data(payload, proof_options)
+    signed = signing_key.sign(hash_data)
+    signature_bytes = signed.signature
+    signature_b64 = base64.b64encode(signature_bytes).decode("ascii")
+    proof = dict(proof_options)
+    proof["proofValue"] = signature_b64
+    secured = dict(payload)
+    secured["proof"] = proof
+    return secured, signature_b64
+
+
+def verify_proof(
+    secured_doc: dict[str, Any],
+    verify_key: nacl.signing.VerifyKey,
+) -> bool:
+    if "proof" not in secured_doc:
+        return False
+    proof = secured_doc["proof"]
+    proof_value_b64 = proof.get("proofValue")
+    if not proof_value_b64:
+        return False
+    proof_options = {k: v for k, v in proof.items() if k != "proofValue"}
+    unsecured = {k: v for k, v in secured_doc.items() if k != "proof"}
+    hash_data = _compute_hash_data(unsecured, proof_options)
+    try:
+        sig = base64.b64decode(proof_value_b64)
+        verify_key.verify(hash_data, sig)
+        return True
+    except Exception:
+        return False
+
+
+def extract_key_id(secured_doc: dict[str, Any]) -> str | None:
+    proof = secured_doc.get("proof") or {}
+    vm = proof.get("verificationMethod") or ""
+    if "/" in vm:
+        return vm.rsplit("/", 1)[-1]
+    return None
+
+
+async def issue_credential(
+    store: PraxisStore,
+    signing_key: nacl.signing.SigningKey | None = None,
+    learner_id: str = "",
+    path: str = "",
+    scenarios_passed: list[str] | None = None,
+    rubric_score: float = 0.0,
+    completed_weeks: int = 6,
+    evidence: list[dict[str, Any]] | None = None,
+    key_id: str | None = None,
+) -> str:
+    if signing_key is None or key_id is None:
+        kp, _enc = await get_active_signing_key(store)
+        signing_key = kp.signing_key
+        key_id = kp.key_id
+    scenarios = list(scenarios_passed or [])
+    ev = list(evidence or [])
+    status_list = BitstringStatusList(store, "default")
+    slot = await status_list.allocate_slot()
+    cred_id = f"vc-{uuid.uuid4().hex[:16]}"
+    payload = build_vc_payload(
+        learner_ref=learner_id,
+        path=path,
+        scenarios_passed=scenarios,
+        rubric_score=rubric_score,
+        completed_weeks=completed_weeks,
+        evidence=ev,
+        credential_id=cred_id,
+        status_list_index=slot,
+    )
+    secured, signature_b64 = sign(payload, signing_key, key_id)
+    payload_json = json.dumps(secured, sort_keys=True, separators=(",", ":"))
+    await store.insert_credential(cred_id, learner_id, payload_json, signature_b64)
+    return cred_id
+
+
+__all__ = [
+    "build_vc_payload",
+    "canonicalize",
+    "sign",
+    "verify_proof",
+    "extract_key_id",
+    "issue_credential",
+    "CREDENTIAL_TIER",
+]

server/vc/issuer_keys.py

@@ -0,0 +1,128 @@
+"""Ed25519 issuer key management (SLICE-09 TASK-09-02).
+
+Private keys are encrypted at rest with nacl.SecretBox using a root key
+from env (D-042). Public keys are stored as base64 strings and served
+publicly for verification. Key rotation = generate new key, mark old
+key as superseded (NOT deleted — old VCs still verify against archived
+public keys).
+"""
+
+from __future__ import annotations
+
+import base64
+import os
+import uuid
+from dataclasses import dataclass
+
+import nacl.secret
+import nacl.signing
+import nacl.utils
+from db.store import PraxisStore
+
+_SECRETBOX_KEY_BYTES = nacl.secret.SecretBox.KEY_SIZE
+
+
+def _load_root_key() -> bytes:
+    raw = os.environ.get("PRAXIS_VC_ISSUER_KEY", "")
+    if raw:
+        kb = raw.encode("utf-8")
+        if len(kb) >= _SECRETBOX_KEY_BYTES:
+            return kb[:_SECRETBOX_KEY_BYTES]
+    return nacl.utils.random(_SECRETBOX_KEY_BYTES)
+
+
+@dataclass
+class KeyPair:
+    key_id: str
+    signing_key: nacl.signing.SigningKey
+    verify_key: nacl.signing.VerifyKey
+    public_key_b64: str
+
+    @property
+    def verification_method(self) -> str:
+        return _verification_method(self.key_id)
+
+
+def _verification_method(key_id: str) -> str:
+    issuer_base = os.environ.get(
+        "PRAXIS_ISSUER_URL", "https://praxis.example/issuers/v0.3"
+    )
+    return f"{issuer_base}/keys/{key_id}"
+
+
+def _encrypt_private_key(signing_key: nacl.signing.SigningKey, root_key: bytes) -> bytes:
+    box = nacl.secret.SecretBox(root_key)
+    nonce = nacl.utils.random(nacl.secret.SecretBox.NONCE_SIZE)
+    ciphertext = box.encrypt(bytes(signing_key), nonce)
+    return ciphertext
+
+
+def _decrypt_private_key(private_key_enc: bytes, root_key: bytes) -> nacl.signing.SigningKey:
+    box = nacl.secret.SecretBox(root_key)
+    seed = box.decrypt(private_key_enc)
+    return nacl.signing.SigningKey(seed)
+
+
+async def init_issuer_key(store: PraxisStore, root_key: bytes | None = None) -> KeyPair:
+    rk = root_key if root_key is not None else _load_root_key()
+    signing_key = nacl.signing.SigningKey.generate()
+    verify_key = signing_key.verify_key
+    public_key_b64 = base64.b64encode(bytes(verify_key)).decode("ascii")
+    private_key_enc = _encrypt_private_key(signing_key, rk)
+    key_id = f"key-{uuid.uuid4().hex[:12]}"
+    await store.init_issuer_key(key_id, public_key_b64, private_key_enc)
+    return KeyPair(key_id, signing_key, verify_key, public_key_b64)
+
+
+async def get_active_signing_key(
+    store: PraxisStore, root_key: bytes | None = None
+) -> tuple[KeyPair, bytes]:
+    rk = root_key if root_key is not None else _load_root_key()
+    row = await store.get_active_signing_key_row()
+    if row is None:
+        kp = await init_issuer_key(store, rk)
+        private_key_enc = await _fetch_private_key_enc(store, kp.key_id)
+        return kp, private_key_enc
+    signing_key = _decrypt_private_key(row["private_key_enc"], rk)
+    verify_key = signing_key.verify_key
+    kp = KeyPair(row["id"], signing_key, verify_key, row["public_key"])
+    return kp, row["private_key_enc"]
+
+
+async def _fetch_private_key_enc(store: PraxisStore, key_id: str) -> bytes:
+    async with store._connect() as db:
+        db.row_factory = None
+        cur = await db.execute(
+            "SELECT private_key_enc FROM issuer_keys WHERE id = ?", (key_id,)
+        )
+        row = await cur.fetchone()
+        return bytes(row[0]) if row else b""
+
+
+async def get_public_key_for_verification(
+    store: PraxisStore, key_id: str
+) -> nacl.signing.VerifyKey:
+    row = await store.get_public_key_row(key_id)
+    if row is None:
+        raise KeyError(f"issuer key {key_id} not found")
+    public_key_bytes = base64.b64decode(row["public_key"])
+    return nacl.signing.VerifyKey(public_key_bytes)
+
+
+async def rotate_key(store: PraxisStore, root_key: bytes | None = None) -> KeyPair:
+    rk = root_key if root_key is not None else _load_root_key()
+    current = await store.get_active_signing_key_row()
+    new_kp = await init_issuer_key(store, rk)
+    if current is not None:
+        await store.set_issuer_key_superseded(current["id"])
+    return new_kp
+
+
+__all__ = [
+    "KeyPair",
+    "init_issuer_key",
+    "get_active_signing_key",
+    "get_public_key_for_verification",
+    "rotate_key",
+    "_verification_method",
+]

server/vc/status_list.py

@@ -0,0 +1,75 @@
+"""Bitstring Status List revocation (SLICE-09 TASK-09-03, REQ-NFR-VC-02).
+
+W3C Bitstring Status List v1.0 — one bit per issued credential. bit=1 means
+revoked. Persisted in SQLite `status_lists` table. Revocation latency = next
+verify call (no cache — status list fetched from SQLite on every verification,
+per REQ-NFR-VC-02). Minimum 131072-bit (16KB) list for herd privacy per spec.
+
+Slot allocation is tracked separately from the revocation bitstring (the
+revocation bit is 0 for a newly-issued active credential, so it cannot
+distinguish "allocated-active" from "never-allocated"). A parallel allocation
+bitstring (`{list_id}_alloc`) records which slots have been handed out.
+"""
+
+from __future__ import annotations
+
+from db.store import PraxisStore
+
+_MIN_BITS = 131072
+
+
+class BitstringStatusList:
+    def __init__(self, store: PraxisStore, list_id: str = "default") -> None:
+        self.store = store
+        self.list_id = list_id
+        self._alloc_id = f"{list_id}_alloc"
+
+    async def _load(self, list_id: str) -> bytearray:
+        row = await self.store.get_status_list(list_id)
+        if row is None:
+            buf = bytearray(_MIN_BITS // 8)
+            await self.store.upsert_status_list(list_id, bytes(buf), _MIN_BITS)
+            return buf
+        return bytearray(row["bitstring"])
+
+    async def set_status(self, credential_idx: int, revoked: bool) -> None:
+        buf = await self._load(self.list_id)
+        byte_pos = credential_idx >> 3
+        bit_pos = credential_idx & 7
+        if revoked:
+            buf[byte_pos] |= 1 << bit_pos
+        else:
+            buf[byte_pos] &= ~(1 << bit_pos)
+        size = len(buf) * 8
+        await self.store.upsert_status_list(self.list_id, bytes(buf), size)
+
+    async def get_status(self, credential_idx: int) -> bool:
+        buf = await self._load(self.list_id)
+        byte_pos = credential_idx >> 3
+        bit_pos = credential_idx & 7
+        if byte_pos >= len(buf):
+            return False
+        return bool((buf[byte_pos] >> bit_pos) & 1)
+
+    async def allocate_slot(self) -> int:
+        buf = await self._load(self._alloc_id)
+        for i in range(len(buf) * 8):
+            byte_pos = i >> 3
+            bit_pos = i & 7
+            if not (buf[byte_pos] >> bit_pos) & 1:
+                buf[byte_pos] |= 1 << bit_pos
+                size = len(buf) * 8
+                await self.store.upsert_status_list(
+                    self._alloc_id, bytes(buf), size
+                )
+                return i
+        new_size = (len(buf) * 8) * 2
+        new_buf = bytearray(new_size // 8)
+        new_buf[: len(buf)] = buf
+        idx = len(buf) * 8
+        new_buf[idx >> 3] |= 1 << (idx & 7)
+        await self.store.upsert_status_list(self._alloc_id, bytes(new_buf), new_size)
+        return idx
+
+
+__all__ = ["BitstringStatusList"]

server/vc/verification.py

@@ -0,0 +1,117 @@
+"""Public VC verification (SLICE-09 TASK-09-04, D-043, REQ-NFR-VC-02).
+
+`GET /vc/verify/<credential_id>` — public, unauthenticated. Fetches the
+credential from SQLite, fetches the issuer public key, validates the Ed25519
+signature against the JCS-canonicalized payload, checks the Bitstring Status
+List (no cache — fetched on every verify call, REQ-NFR-VC-02). Returns JSON
+{valid, status, issuer, credential, mastery, credentialTier, verifiedAt}.
+No PII beyond what the credential asserts.
+"""
+
+from __future__ import annotations
+
+import datetime as _dt
+import json
+from typing import Any
+
+from db.store import PraxisStore
+
+from server.vc.issuer import verify_proof, extract_key_id, CREDENTIAL_TIER
+from server.vc.issuer_keys import get_public_key_for_verification
+from server.vc.status_list import BitstringStatusList
+
+
+def _now_iso() -> str:
+    return _dt.datetime.now(_dt.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
+
+
+async def verify_credential(
+    store: PraxisStore, credential_id: str
+) -> dict[str, Any] | None:
+    row = await store.get_credential(credential_id)
+    if row is None:
+        return None
+    secured_doc = json.loads(row["vc_payload_json"])
+    key_id = extract_key_id(secured_doc)
+    if key_id is None:
+        return _invalid(row, secured_doc)
+    try:
+        verify_key = await get_public_key_for_verification(store, key_id)
+    except KeyError:
+        return _invalid(row, secured_doc)
+    sig_valid = verify_proof(secured_doc, verify_key)
+    revoked = False
+    cs = secured_doc.get("credentialStatus") or {}
+    idx_str = cs.get("statusListIndex")
+    if idx_str is not None:
+        sl = BitstringStatusList(store, "default")
+        revoked = await sl.get_status(int(idx_str))
+    status = "revoked" if revoked else "active"
+    valid = bool(sig_valid and not revoked)
+    subject = secured_doc.get("credentialSubject") or {}
+    issuer = secured_doc.get("issuer")
+    return {
+        "valid": valid,
+        "status": status,
+        "issuer": issuer,
+        "credential": {
+            "id": secured_doc.get("id"),
+            "type": secured_doc.get("type"),
+            "validFrom": secured_doc.get("validFrom"),
+            "validUntil": secured_doc.get("validUntil"),
+        },
+        "mastery": {
+            "skill": subject.get("skill"),
+            "level": subject.get("level"),
+            "path": subject.get("path"),
+            "rubricScore": subject.get("rubricScore"),
+            "scenariosPassed": subject.get("scenariosPassed", []),
+            "completedWeeks": subject.get("completedWeeks"),
+        },
+        "credentialTier": subject.get("credentialTier", CREDENTIAL_TIER),
+        "verifiedAt": _now_iso(),
+    }
+
+
+def _invalid(row: dict, secured_doc: dict) -> dict[str, Any]:
+    subject = secured_doc.get("credentialSubject") or {}
+    return {
+        "valid": False,
+        "status": row.get("status", "active"),
+        "issuer": secured_doc.get("issuer"),
+        "credential": {
+            "id": secured_doc.get("id"),
+            "type": secured_doc.get("type"),
+            "validFrom": secured_doc.get("validFrom"),
+            "validUntil": secured_doc.get("validUntil"),
+        },
+        "mastery": {
+            "skill": subject.get("skill"),
+            "level": subject.get("level"),
+            "path": subject.get("path"),
+            "rubricScore": subject.get("rubricScore"),
+            "scenariosPassed": subject.get("scenariosPassed", []),
+            "completedWeeks": subject.get("completedWeeks"),
+        },
+        "credentialTier": subject.get("credentialTier", CREDENTIAL_TIER),
+        "verifiedAt": _now_iso(),
+    }
+
+
+async def revoke_credential(store: PraxisStore, credential_id: str) -> bool:
+    row = await store.get_credential(credential_id)
+    if row is None:
+        return False
+    secured_doc = json.loads(row["vc_payload_json"])
+    cs = secured_doc.get("credentialStatus") or {}
+    idx_str = cs.get("statusListIndex")
+    if idx_str is None:
+        await store.set_credential_status(credential_id, "revoked")
+        return True
+    sl = BitstringStatusList(store, "default")
+    await sl.set_status(int(idx_str), True)
+    await store.set_credential_status(credential_id, "revoked")
+    return True
+
+
+__all__ = ["verify_credential", "revoke_credential"]