Spaces:

asnannp
/

DocDoeAI

Runtime error

DocDoeAI / app /services /mastery_engine.py
deploy: sync backend to Space root (learn-lesson HF cache fix)
65d68f5 about 1 month ago
28.7 kB
"""Evidence-based mastery engine — the single place mastery math lives.
Pure, deterministic functions only (no DB, no AI, no randomness) so every rule
is unit-testable and inspectable. Persistence stays in the EXISTING
``topic_mastery`` row: the numeric columns plus a versioned dict stored in its
``evidence`` JSON column (no schema migration needed). Callers
(``adaptive_engine``) load/store rows; this module only computes.
Documented policies
-------------------
Evidence strength (0..1). Different actions are different proof:
- finishing a lesson/video is weak evidence (0.15-0.20)
- answering a recall prompt correctly is stronger (0.30)
- a correct MCQ checkpoint is stronger (0.45)
- a correct short/board answer is stronger (0.50-0.60)
- solving a numerical is strongest (0.70)
- a hint halves the strength of a correct answer
- re-answering a question already answered correctly earns 25% strength
  (memorising one question must not prove mastery)
Score: exponentially weighted toward the new result, where the learning rate
scales with evidence strength (strong evidence moves the score more, weak
evidence barely moves it). Wrong answers weigh 1.25x their strength so a
recent failure genuinely reduces standing.
Revision intervals (deterministic, not a memory-science claim):
- needs_repair -> 1 day, developing -> 2 days
- secure with confidence < 0.6 -> 4 days, secure otherwise -> 7 days
- each consecutive successful recall beyond 2 multiplies the interval by 1.5,
  capped at 21 days; always clamped to finish at least 1 day before the exam.
"""

from __future__ import annotations

from dataclasses import dataclass, field
from datetime import date, datetime, timedelta
import logging
import re as _re
from typing import Any, Literal

logger = logging.getLogger(__name__)

MASTERY_ENGINE_VERSION = 2

EvidenceKind = Literal[
    "lesson_completed",
    "video_completed",
    "recall_correct",
    "checkpoint_mcq",
    "checkpoint_short",
    "checkpoint_numerical",
    "checkpoint_diagram",
    "board_answer",
    "transfer_check",
    "revision_recall",
    "quiz_aggregate",
]

EVIDENCE_STRENGTH: dict[str, float] = {
    "lesson_completed": 0.20,
    "video_completed": 0.15,
    "recall_correct": 0.30,
    "checkpoint_mcq": 0.45,
    "checkpoint_short": 0.50,
    "checkpoint_numerical": 0.70,
    "checkpoint_diagram": 0.50,
    "board_answer": 0.60,
    "transfer_check": 0.55,
    "revision_recall": 0.50,
    "quiz_aggregate": 0.50,
}

# Evidence kinds that count as an actual understanding check (a completed
# lesson is exposure, not proof).
CHECK_EVIDENCE_KINDS: frozenset[str] = frozenset(
    kind for kind in EVIDENCE_STRENGTH if kind not in {"lesson_completed", "video_completed"}
)

HINT_STRENGTH_MULTIPLIER = 0.5
REPEAT_QUESTION_MULTIPLIER = 0.25
WRONG_ANSWER_WEIGHT = 1.25
BASE_LEARNING_RATE = 0.6
MIN_LEARNING_RATE = 0.05
MAX_LEARNING_RATE = 0.5

SECURE_SCORE = 75.0
SECURE_CONFIDENCE = 0.5
SECURE_CONSECUTIVE = 2
AT_RISK_OVERDUE_DAYS = 4
AT_RISK_EXAM_WINDOW_DAYS = 14
AT_RISK_SCORE = 60.0

MAX_LOG_ENTRIES = 20
MAX_SEEN_QUESTIONS = 40

MasteryState = Literal[
    "not_started",
    "introduced",
    "developing",
    "needs_repair",
    "secure",
    "revision_due",
    "at_risk",
]

ErrorCategory = Literal[
    "not_attempted",
    "concept_misunderstanding",
    "formula_selection",
    "unit_error",
    "numerical_error",
    "sign_or_direction_error",
    "calculation_error",
    "diagram_interpretation",
    "incomplete_explanation",
    "missing_exam_keyword",
    "memorized_answer",
]


@dataclass
class EvidenceEvent:
    kind: str
    correct: bool
    at: datetime
    question_id: str | None = None
    hint_used: bool = False
    attempt_index: int = 1
    error_category: str | None = None
    time_spent_seconds: int | None = None
    source: str = "guided_class"


@dataclass
class MasterySnapshot:
    """The mutable mastery values for one (student, concept)."""

    score: float = 0.0
    confidence: float = 0.0
    attempts_count: int = 0
    state: str = "not_started"
    next_review_at: datetime | None = None
    evidence: dict[str, Any] = field(default_factory=dict)


@dataclass
class MasteryUpdate:
    before_score: float
    after_score: float
    before_state: str
    after_state: str
    confidence: float
    consecutive_success: int
    applied_strength: float
    next_review_at: datetime | None
    evidence: dict[str, Any]
    attempts_count: int
    # D1: whether THIS event was credited toward the consecutive-success
    # streak. An already-correct repeated question never earns new streak
    # credit, so it cannot farm a concept toward ``secure`` nor extend the
    # review interval. (Score/evidence dampening is preserved separately.)
    streak_credited: bool = False
    # Base mastery state beneath any urgency overlay. Overlays
    # (needs_repair / at_risk / revision_due) are not mastery improvements
    # and must never be counted as gains in M7.
    before_base_state: str = "not_started"
    after_base_state: str = "not_started"
    # Resolved urgency overlay, if any, after this event (None == clear).
    after_overlay: str | None = None


# Canonical mastery model (Product Strategy v1.1, §6.2). The four base
# states are the ONLY mastery progression; the three overlays are urgency
# flags that sit on top of a base state and whose clearing is explicitly
# NOT a mastery gain.
BASE_MASTERY_STATES: tuple[str, ...] = (
    "not_started",
    "introduced",
    "developing",
    "secure",
)
URGENCY_OVERLAYS: tuple[str, ...] = (
    "needs_repair",
    "at_risk",
    "revision_due",
)


def base_mastery_state(resolved_state: str) -> str:
    """Collapse a resolved state (which may carry an urgency overlay) to the
    base mastery state beneath it.
    Overlays sit on top of a base state. ``needs_repair`` and ``at_risk`` sit
    over ``developing`` (or rarely ``secure``); ``revision_due`` sits over
    ``developing`` or ``secure``. The corrected metrics spec requires that an
    overlay clearing is never reported as a mastery gain, so any analytics
    measuring learning progress must call this and ignore overlays.
    """
    if resolved_state in BASE_MASTERY_STATES:
        return resolved_state
    # Overlays always sit over a non-secure, non-introduced base: the concept
    # has check evidence (it was answered) and is not yet secure. This is the
    # honest conservative reading — an overlay never hides a ``secure`` claim
    # because ``compute_state`` already gates secure before the overlays.
    return "developing"


def compute_base_state(
    *,
    score: float,
    confidence: float,
    consecutive_success: int,
    attempts_count: int,
    has_check_evidence: bool,
) -> MasteryState:
    """Deterministic base mastery state, ignoring all urgency overlays.
    Uses the SAME thresholds as ``compute_state`` (no new [HYPOTHESIS - TUNE]
    constants are introduced); it simply does not evaluate the overlay
    branches. This is the value M7 measures movement on.
    """
    if attempts_count == 0:
        return "not_started"
    if not has_check_evidence:
        return "introduced"
    if score >= SECURE_SCORE and confidence >= SECURE_CONFIDENCE and consecutive_success >= SECURE_CONSECUTIVE:
        return "secure"
    return "developing"


def overlay_for(resolved_state: str) -> str | None:
    """Return the urgency overlay name, or None when the state is a pure base."""
    return resolved_state if resolved_state in URGENCY_OVERLAYS else None


def _fresh_evidence() -> dict[str, Any]:
    return {
        "version": MASTERY_ENGINE_VERSION,
        "consecutive_success": 0,
        "hint_count": 0,
        "first_attempt_correct": 0,
        "first_attempt_total": 0,
        "error_categories": {},
        "seen_question_correct": [],
        "log": [],
        "last_correct_at": None,
        "last_wrong_at": None,
    }


def _normalized_evidence(raw: dict[str, Any] | None) -> dict[str, Any]:
    data = _fresh_evidence()
    if isinstance(raw, dict):
        for key in data:
            if key in raw and isinstance(raw[key], type(data[key]) if data[key] is not None else object):
                data[key] = raw[key]
        # tolerate legacy v1 payloads ({"last_score_percent": ...}) silently
    return data


def independent_success_streak(evidence: dict[str, Any]) -> int:
    """Reconstruct independent success, including logs written by older code.
    Legacy logs may have credited hinted answers. They must not continue to
    prove secure mastery when an existing student returns or answers again.
    """
    log = evidence.get("log")
    if not isinstance(log, list) or not log:
        return max(0, int(evidence.get("consecutive_success", 0) or 0))
    streak = 0
    seen_correct: set[str] = set()
    for entry in log:
        if not isinstance(entry, dict) or entry.get("kind") not in CHECK_EVIDENCE_KINDS:
            continue
        question_id = entry.get("question_id")
        repeat = bool(question_id) and question_id in seen_correct
        if entry.get("correct") and isinstance(question_id, str):
            seen_correct.add(question_id)
        if not entry.get("correct") or entry.get("hint_used") or entry.get("attempt_index", 1) != 1:
            streak = 0
            continue
        strength = entry.get("strength")
        # Older repeated answers have no streak marker, but their dampened
        # strength distinguishes them even after a bounded log is truncated.
        dampened = isinstance(strength, (int, float)) and strength < EVIDENCE_STRENGTH[entry["kind"]]
        if repeat or entry.get("streak_credited") is False or dampened:
            continue
        streak += 1
    return streak


def effective_strength(event: EvidenceEvent, evidence: dict[str, Any]) -> float:
    """Strength of THIS event given history (hints, repeated questions)."""
    strength = EVIDENCE_STRENGTH.get(event.kind, 0.3)
    if event.hint_used and event.correct:
        strength *= HINT_STRENGTH_MULTIPLIER
    if (
        event.correct
        and event.question_id
        and event.question_id in evidence.get("seen_question_correct", [])
    ):
        strength *= REPEAT_QUESTION_MULTIPLIER
    return round(strength, 4)


def compute_state(
    *,
    score: float,
    confidence: float,
    consecutive_success: int,
    attempts_count: int,
    has_check_evidence: bool,
    has_open_repair: bool,
    next_review_at: datetime | None,
    now: datetime,
    exam_date: date | None,
) -> MasteryState:
    """Deterministic state resolution, priority top-down."""
    if attempts_count == 0:
        return "not_started"
    if has_open_repair:
        return "needs_repair"
    if next_review_at is not None and now >= next_review_at:
        overdue_days = (now - next_review_at).days
        exam_close = (
            exam_date is not None
            and (exam_date - now.date()).days <= AT_RISK_EXAM_WINDOW_DAYS
        )
        if overdue_days > AT_RISK_OVERDUE_DAYS or (exam_close and score < AT_RISK_SCORE):
            return "at_risk"
        return "revision_due"
    if not has_check_evidence:
        return "introduced"
    if score >= SECURE_SCORE and confidence >= SECURE_CONFIDENCE and consecutive_success >= SECURE_CONSECUTIVE:
        return "secure"
    return "developing"


def review_interval_days(state: str, confidence: float, consecutive_success: int) -> int:
    if state == "needs_repair":
        return 1  # repairs revise tomorrow, no streak credit
    if state in {"developing", "introduced"}:
        return 2
    base = 4 if confidence < 0.6 else 7
    extra_recalls = max(0, consecutive_success - SECURE_CONSECUTIVE)
    interval = base * (1.5 ** extra_recalls)
    return min(21, max(1, round(interval)))


def clamp_review_to_exam(review_at: datetime, exam_date: date | None, now: datetime) -> datetime:
    if exam_date is None:
        return review_at
    latest = datetime.combine(exam_date - timedelta(days=1), datetime.min.time(), tzinfo=review_at.tzinfo)
    if review_at > latest:
        return max(now + timedelta(days=1), latest)
    return review_at


def apply_evidence(
    snapshot: MasterySnapshot,
    event: EvidenceEvent,
    *,
    exam_date: date | None,
    has_open_repair: bool,
) -> MasteryUpdate:
    """Apply one evidence event to a mastery snapshot (mutates the snapshot)."""
    evidence = _normalized_evidence(snapshot.evidence)
    evidence["consecutive_success"] = independent_success_streak(evidence)
    before_score = snapshot.score
    before_state = snapshot.state or "not_started"

    strength = effective_strength(event, evidence)
    target = 100.0 if event.correct else 0.0
    weight = strength * (1.0 if event.correct else WRONG_ANSWER_WEIGHT)
    learning_rate = max(MIN_LEARNING_RATE, min(MAX_LEARNING_RATE, weight * BASE_LEARNING_RATE))
    if snapshot.attempts_count == 0:
        # First evidence anchors the score instead of averaging against 0.
        after_score = target * max(strength, 0.2)
    else:
        after_score = snapshot.score * (1 - learning_rate) + target * learning_rate
    after_score = round(max(0.0, min(100.0, after_score)), 2)

    if event.correct:
        snapshot.confidence = round(min(1.0, snapshot.confidence + strength * 0.15), 4)
    else:
        snapshot.confidence = round(max(0.05, snapshot.confidence - 0.2), 4)

    is_check = event.kind in CHECK_EVIDENCE_KINDS
    # D1: a repeat of an already-correct question must NOT earn new
    # consecutive-success credit. Without this guard a student can farm one
    # easy question into ``secure`` (satisfying SECURE_CONSECUTIVE) and
    # multiply the review interval. Score/evidence dampening is preserved by
    # ``effective_strength`` above; only the streak counter is gated here.
    seen_correct_before = evidence.get("seen_question_correct", [])
    is_repeat_of_correct = bool(event.question_id) and event.question_id in seen_correct_before
    streak_credited = False
    if is_check:
        if event.correct:
            if event.hint_used or event.attempt_index != 1:
                evidence["consecutive_success"] = 0
            elif not is_repeat_of_correct:
                evidence["consecutive_success"] = int(evidence.get("consecutive_success", 0)) + 1
                streak_credited = True
            evidence["last_correct_at"] = event.at.isoformat()
        else:
            evidence["consecutive_success"] = 0
            evidence["last_wrong_at"] = event.at.isoformat()
        if event.attempt_index == 1 and not event.hint_used and not is_repeat_of_correct:
            evidence["first_attempt_total"] = int(evidence.get("first_attempt_total", 0)) + 1
            if event.correct:
                evidence["first_attempt_correct"] = int(evidence.get("first_attempt_correct", 0)) + 1

    if event.hint_used:
        evidence["hint_count"] = int(evidence.get("hint_count", 0)) + 1
    if not event.correct and event.error_category:
        categories = dict(evidence.get("error_categories", {}))
        categories[event.error_category] = int(categories.get(event.error_category, 0)) + 1
        evidence["error_categories"] = categories
    if event.correct and event.question_id:
        seen = list(evidence.get("seen_question_correct", []))
        if event.question_id not in seen:
            seen.append(event.question_id)
        evidence["seen_question_correct"] = seen[-MAX_SEEN_QUESTIONS:]

    log = list(evidence.get("log", []))
    log.append(
        {
            "kind": event.kind,
            "correct": event.correct,
            "strength": strength,
            "at": event.at.isoformat(),
            "question_id": event.question_id,
            "error_category": event.error_category if not event.correct else None,
            "hint_used": event.hint_used,
            "attempt_index": event.attempt_index,
            "streak_credited": streak_credited,
            "source": event.source,
        }
    )
    evidence["log"] = log[-MAX_LOG_ENTRIES:]

    snapshot.score = after_score
    snapshot.attempts_count += 1

    consecutive = int(evidence.get("consecutive_success", 0))
    has_check_evidence = int(evidence.get("first_attempt_total", 0)) > 0 or any(
        item.get("kind") in CHECK_EVIDENCE_KINDS for item in log if isinstance(item, dict)
    )

    # Successful evidence schedules the next recall; failures schedule sooner.
    provisional_state = compute_state(
        score=snapshot.score,
        confidence=snapshot.confidence,
        consecutive_success=consecutive,
        attempts_count=snapshot.attempts_count,
        has_check_evidence=has_check_evidence,
        has_open_repair=has_open_repair,
        next_review_at=None,  # review scheduling below uses the fresh state
        now=event.at,
        exam_date=exam_date,
    )
    interval = review_interval_days(provisional_state, snapshot.confidence, consecutive)
    snapshot.next_review_at = clamp_review_to_exam(
        event.at + timedelta(days=interval), exam_date, event.at
    )

    after_state = provisional_state
    snapshot.state = after_state
    evidence["version"] = MASTERY_ENGINE_VERSION
    snapshot.evidence = evidence
    logger.debug(
        "[FIX:independent-mastery] evidence applied",
        extra={"evidence_kind": event.kind, "assisted": event.hint_used or event.attempt_index != 1,
               "streak_credited": streak_credited, "mastery_state": after_state},
    )

    after_base = compute_base_state(
        score=snapshot.score,
        confidence=snapshot.confidence,
        consecutive_success=consecutive,
        attempts_count=snapshot.attempts_count,
        has_check_evidence=has_check_evidence,
    )
    before_base = base_mastery_state(before_state)

    return MasteryUpdate(
        before_score=round(before_score, 2),
        after_score=snapshot.score,
        before_state=before_state,
        after_state=after_state,
        confidence=snapshot.confidence,
        consecutive_success=consecutive,
        applied_strength=strength,
        next_review_at=snapshot.next_review_at,
        evidence=evidence,
        attempts_count=snapshot.attempts_count,
        streak_credited=streak_credited,
        before_base_state=before_base,
        after_base_state=after_base,
        after_overlay=overlay_for(after_state),
    )


# ---------------------------------------------------------------------------
# Deterministic error classification
# ---------------------------------------------------------------------------

_NUMBER_PATTERN = r"-?\d+(?:[.,]\d+)?"

# Multi-character unit tokens recognised on a word boundary. Single-letter
# units (m, s, g, k, n, j, w) are only credited when they attach directly to a
# number (e.g. "340 m", "340m"), never as a free word — otherwise ordinary
# prose ("my answer is five") would be misread as carrying a unit.
_UNIT_WORD_TOKENS = (
    "m/s", "ms-1", "m/s2", "m/s^2", "hz", "khz", "mhz",
    "hertz", "metre", "metres", "meter", "meters", "second", "seconds",
    "sec", "secs", "minute", "minutes", "cm", "km", "mm", "nm", "kg",
    "ms", "mol", "kelvin", "ampere", "amp", "amps", "volt", "volts",
    "ohm", "watt", "watts", "joule", "joules", "newton", "newtons",
    "pascal", "rad", "radian", "degree", "deg",
)
# Single-letter units credited only when immediately adjacent to a number.
_UNIT_ADJACENT_LETTERS = ("m", "s", "g", "k", "n", "j", "w", "a", "v")


def _extract_numbers(text: str) -> list[float]:
    values: list[float] = []
    for raw in _re.findall(_NUMBER_PATTERN, text.replace(",", ".")):
        try:
            values.append(float(raw))
        except ValueError:
            continue
    return values


def _detect_units(text: str) -> frozenset[str]:
    """Reliably detect physical units in an answer.
    This replaces the old substring ``_contains_unit`` which falsely fired on
    ordinary prose (``" my answer is 5 "`` contains ``" m"`` and was treated
    as a metre unit). Word-boundary multi-char tokens plus a number-adjacency
    rule for single letters makes unit detection reliable enough to separate
    ``unit_error`` from ``calculation_error`` ONLY in the high-confidence
    power-of-ten case; everything ambiguous degrades to ``numerical_error``.
    """
    if not text:
        return frozenset()
    lowered = text.casefold()
    detected: set[str] = set()
    for unit in _UNIT_WORD_TOKENS:
        if _re.search(rf"\b{_re.escape(unit)}\b", lowered):
            detected.add(unit)
    # Single-letter units count only when stuck to a number ("340 m", "340m",
    # "5s") — never as a free word in prose.
    for match in _re.finditer(r"-?\d+(?:[.,]\d+)?\s*([a-z])", lowered):
        letter = match.group(1)
        if letter in _UNIT_ADJACENT_LETTERS:
            detected.add(letter)
    return frozenset(detected)


def _keyword_coverage(answer: str, keywords: list[str]) -> tuple[int, int]:
    lowered = answer.casefold()
    matched = sum(1 for keyword in keywords if keyword.casefold() in lowered)
    return matched, len(keywords)


def _power_of_ten_ratio(student: float, expected: float) -> bool:
    """True when the student value is exactly a 10/100/1000 (or inverse)
    multiple of the expected value — the most defensible unit-conversion
    signal because it needs no text parsing."""
    if expected == 0:
        return False
    return abs(student / expected) in (10.0, 100.0, 1000.0, 0.1, 0.01, 0.001)


def classify_error(
    *,
    question_type: str,
    student_answer: str,
    correct_answer: str,
    expected_keywords: list[str] | None = None,
    misconception: str | None = None,
) -> ErrorCategory:
    """Deterministic first-pass error classification.
    Not an AI judgement — a small, inspectable rule set. AI may LATER refine a
    category, but the stored category never comes from an unvalidated model.
    v1.1 graceful-degradation rule (Product Strategy §15.3): where the
    deterministic rules cannot *reliably* separate ``unit_error`` from
    ``calculation_error`` from a final answer string alone, the distinction is
    not shipped. The unreliable cases merge into the broader
    ``numerical_error`` bucket, which shares the ``scaffolded_numerical``
    remediation with both — so a misclassification never sends the wrong
    lesson. The feature degrades; the architecture (no LLM) does not.
    """
    answer = (student_answer or "").strip()
    keywords = expected_keywords or []

    # D2: a blank/whitespace answer is NOT evidence of a physics
    # misunderstanding. It is classified first, before every other rule, so a
    # student who submits nothing is never told their concept is broken.
    if not answer:
        return "not_attempted"

    if question_type == "diagram":
        return "diagram_interpretation"

    if question_type == "numerical":
        expected_numbers = _extract_numbers(correct_answer)
        student_numbers = _extract_numbers(answer)
        if expected_numbers and student_numbers:
            # Exact magnitude match -> the arithmetic happened; the question
            # is whether a unit is missing/mismatched.
            magnitude_match = any(
                abs(sn - en) < 1e-6 for sn in student_numbers for en in expected_numbers
            )
            if magnitude_match:
                expected_units = _detect_units(correct_answer)
                student_units = _detect_units(answer)
                if expected_units and not student_units:
                    # Right number, no unit written -> reliably a unit error.
                    return "unit_error"
                # Right number AND a unit is present but still marked wrong is
                # ambiguous (could be a unit-format rubric, a mis-grade, or a
                # sign). Degrade rather than guess calculation_error.
                return "numerical_error"
            # Off by an exact power of ten -> almost always a unit-conversion
            # slip. Deterministic, no text parsing, high confidence.
            if any(_power_of_ten_ratio(sn, en) for sn in student_numbers for en in expected_numbers):
                return "unit_error"
            # A wrong number with no reliable unit signal degrades to the
            # broader numerical bucket. The old code branched on a
            # chapter-specific ``_FORMULA_TOKENS`` list (Sound Waves only) and
            # a fragile substring unit check; both misfired past chapter 1.
            return "numerical_error"
        if not student_numbers:
            # The student wrote no number at all for a numerical -> never
            # reached a substitution. This is the one reliably separable
            # ``formula_selection`` signal.
            return "formula_selection"
        return "numerical_error"

    if question_type == "mcq":
        return "concept_misunderstanding"

    # short / board-style answers
    matched, total = _keyword_coverage(answer, keywords)
    if total > 0 and matched == 0:
        return "concept_misunderstanding"
    if total > 0 and matched < total:
        return "missing_exam_keyword"
    if len(answer.split()) < 6:
        return "incomplete_explanation"
    return "incomplete_explanation"


# ---------------------------------------------------------------------------
# Student-facing copy per error category (deterministic diagnosis + activity)
# ---------------------------------------------------------------------------

ERROR_CATEGORY_COPY: dict[str, dict[str, str]] = {
    "not_attempted": {
        "diagnosis": "No answer was submitted this time — there is no evidence of a misunderstanding yet.",
        "activity": "scaffolded_first_step",
        "activity_prompt": "Open the first step of this question (the setup is scaffolded) and try just that part, then check it.",
    },
    "concept_misunderstanding": {
        "diagnosis": "The core idea itself is not settled yet — this was not a careless slip.",
        "activity": "re_explain",
        "activity_prompt": "Reread the one explanation block for this idea, then answer one fresh check question in your own words.",
    },
    "formula_selection": {
        "diagnosis": "The wrong formula (or no formula) was chosen for this situation.",
        "activity": "worked_solution",
        "activity_prompt": "Compare the formulas you know for this chapter and match each to the situation it serves, then redo one worked example.",
    },
    "unit_error": {
        "diagnosis": "The physics was right but the units were not converted or written.",
        "activity": "scaffolded_numerical",
        "activity_prompt": "Solve one scaffolded numerical where the first step is converting every value to SI units before substituting.",
    },
    "sign_or_direction_error": {
        "diagnosis": "A direction or sign flipped somewhere in the working.",
        "activity": "worked_solution",
        "activity_prompt": "Correct one worked solution line-by-line, marking where the direction changed.",
    },
    "calculation_error": {
        "diagnosis": "Setup was right; the arithmetic slipped.",
        "activity": "scaffolded_numerical",
        "activity_prompt": "Redo the same style of numerical slowly, writing every substitution step.",
    },
    "numerical_error": {
        "diagnosis": "Something in the numerical working went wrong — the setup, a substitution, or the units. We can't tell which from the final number alone, so we won't guess.",
        "activity": "scaffolded_numerical",
        "activity_prompt": "Redo the numerical step-by-step, converting every value to SI units first and writing each substitution before solving.",
    },
    "diagram_interpretation": {
        "diagnosis": "The diagram's labels or meaning were misread.",
        "activity": "label_diagram",
        "activity_prompt": "Label the key parts of this concept's diagram from memory, then check against the board.",
    },
    "incomplete_explanation": {
        "diagnosis": "The idea is present but the answer stops before the marks are earned.",
        "activity": "board_answer_rewrite",
        "activity_prompt": "Rewrite the answer using the full answer structure — definition, reason, and conclusion.",
    },
    "missing_exam_keyword": {
        "diagnosis": "The answer misses the exact keywords the board awards marks for.",
        "activity": "board_answer_rewrite",
        "activity_prompt": "Rewrite one board answer including every underlined keyword.",
    },
    "memorized_answer": {
        "diagnosis": "The same memorised line is repeating without transfer to new questions.",
        "activity": "transfer_question",
        "activity_prompt": "Answer one unseen question on the same idea, phrased differently.",
    },
}