"""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.", }, }