Spaces:

asnannp
/

DocDoeAI

Sleeping

DocDoeAI / app /services /adaptive_engine.py
deploy: sync backend to Space root (learn-lesson HF cache fix)
ab01cae 27 days ago
33.8 kB
"""Adaptive engine orchestration: checkpoints, repair queue, daily planning.
DB-facing layer over the pure ``mastery_engine``. Every mutation is
ownership-scoped, idempotent where the student can double-submit, and stores a
human-readable reason so the product can always answer
"DocDoe chose this task because…".
"""

from __future__ import annotations

import logging
from datetime import datetime, timezone

from fastapi import HTTPException, status
from sqlalchemy import select
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session

from app.models.learning_state import (
    Chapter,
    CheckpointEvent,
    DailyTask,
    RepairItem,
    RevisionItem,
    StudentProfileState,
    Subject,
    TopicMastery,
    UsageEvent,
)
from app.schemas.learning_state import (
    CheckpointConsequenceResponse,
    CheckpointMasteryOut,
    CheckpointRequest,
    PlanTodayResponse,
    RepairItemOut,
)
from app.services import concept_graph
from app.services.learning_events import (
    EVENT_CHECKPOINT_RECORDED,
    EVENT_HINT_USED,
    EVENT_REPAIR_CREATED,
    EVENT_REPAIR_ESCALATED,
    EVENT_REPAIR_RESOLVED,
    EVENT_REPAIR_TASK_OPENED,
    add_learning_event,
    record_mastery_transition,
)
from app.services.mastery_engine import (
    ERROR_CATEGORY_COPY,
    MASTERY_ENGINE_VERSION,
    EvidenceEvent,
    MasterySnapshot,
    apply_evidence,
    base_mastery_state,
    classify_error,
    overlay_for,
)
from app.services.today_plan_service import refresh_daily_plan_totals, replan_today
from app.services.planning_context import aware
from app.services.checkpoint_evaluation import EVALUATION_VERSION, evaluate_checkpoint


REPAIR_TASK_MINUTES = 10
logger = logging.getLogger(__name__)


def _now() -> datetime:
    return datetime.now(timezone.utc)


def _owned_chapter_for_checkpoint(
    db: Session,
    *,
    user_id: str,
    payload: CheckpointRequest,
) -> Chapter | None:
    if payload.chapter_id:
        chapter = db.scalar(
            select(Chapter).where(
                Chapter.id == payload.chapter_id,
                Chapter.user_id == user_id,
            )
        )
        if chapter is None:
            raise HTTPException(404, detail="Chapter not found.")
        if chapter.catalog_id != payload.chapter_catalog_id:
            raise HTTPException(422, detail="Checkpoint chapter does not match its published question.")
        if payload.subject_id and payload.subject_id != chapter.subject_id:
            raise HTTPException(422, detail="Checkpoint subject does not match its chapter.")
        return chapter
    if payload.subject_id and db.scalar(select(Subject.id).where(
        Subject.id == payload.subject_id, Subject.user_id == user_id,
    )) is None:
        raise HTTPException(404, detail="Subject not found.")
    return db.scalar(
        select(Chapter).where(
            Chapter.user_id == user_id,
            Chapter.catalog_id == payload.chapter_catalog_id,
        )
    )


def _sync_revision_item_from_checkpoint(
    db: Session,
    *,
    user_id: str,
    payload: CheckpointRequest,
    chapter: Chapter | None,
    next_review_at: datetime | None,
    now: datetime,
    correct: bool,
) -> None:
    if next_review_at is None:
        return
    client_item_id = f"mastery-review:{payload.concept_key}"[:180]
    item = db.scalar(
        select(RevisionItem).where(
            RevisionItem.user_id == user_id,
            RevisionItem.client_item_id == client_item_id,
        ).with_for_update()
    )
    due_status = "due" if next_review_at.date() <= now.date() else "pending"
    if item is None:
        db.add(
            RevisionItem(
                user_id=user_id,
                client_item_id=client_item_id,
                subject_id=(chapter.subject_id if chapter else None) or payload.subject_id,
                chapter_id=(chapter.id if chapter else None) or payload.chapter_id,
                mission_id=payload.mission_id,
                topic_key=payload.concept_key,
                title=payload.concept_label,
                source_kind="checkpoint",
                source_ref=payload.question_id,
                status=due_status,
                due_at=next_review_at,
                item_data={
                    "chapter_catalog_id": payload.chapter_catalog_id,
                    "estimated_minutes": 10,
                },
            )
        )
        return
    if item.status == "dismissed":
        # A student-controlled dismissal is not evidence of mastery, but it is
        # still an explicit choice. Checkpoints must not undo that choice.
        return
    if item.status in {"completed", "resolved"}:
        if correct:
            return
        # A finished recall cycle is not a lifetime guarantee. Fresh failure
        # restarts the active spacing cycle, without deleting past evaluations
        # or the append-only ledger that explains the earlier completion.
        metadata = dict(item.item_data or {})
        previous_status = item.status
        previous_successes = int(metadata.get("successful_reviews", 0) or 0)
        add_learning_event(
            db,
            user_id=user_id,
            event_type="REVISION_REACTIVATED",
            entity_type="revision_item",
            entity_id=item.id,
            idempotency_key=f"revision-reactivated:{payload.client_event_id}",
            subject_id=item.subject_id,
            chapter_id=item.chapter_id,
            topic_key=item.topic_key,
            event_data={
                "reason": "new_incorrect_checkpoint",
                "previous_status": previous_status,
                "previous_successful_reviews": previous_successes,
                "previous_completed_at": (
                    item.completed_at.isoformat() if item.completed_at else None
                ),
                "next_due_at": next_review_at.isoformat(),
                "checkpoint_client_event_id": payload.client_event_id,
            },
        )
        item.item_data = {**metadata, "successful_reviews": 0}
        item.completed_at = None
        logger.debug(
            "[FIX:revision-reactivation] restarted completed recall cycle "
            "previous_status=%s previous_successes=%d",
            previous_status,
            previous_successes,
        )
    item.due_at = next_review_at
    item.status = due_status
    item.mission_id = item.mission_id or payload.mission_id
    item.chapter_id = item.chapter_id or (chapter.id if chapter else payload.chapter_id)
    item.subject_id = item.subject_id or (chapter.subject_id if chapter else payload.subject_id)


def repair_item_out(item: RepairItem) -> RepairItemOut:
    return RepairItemOut(
        id=item.id,
        subject_id=item.subject_id,
        chapter_id=item.chapter_id,
        concept_key=item.concept_key,
        concept_label=item.concept_label,
        mission_id=item.mission_id,
        error_category=item.error_category,
        diagnosis=item.diagnosis,
        recommended_activity=item.recommended_activity,
        activity_prompt=item.activity_prompt,
        priority=item.priority,
        estimated_minutes=item.estimated_minutes,
        status=item.status,
        support_level=item.support_level,
        failed_attempts=item.failed_attempts,
        retry_result=item.retry_result,
        mastery_recovered=item.mastery_recovered,
        created_at=item.created_at,
        resolved_at=item.resolved_at,
    )


def _open_repairs_for_concept(db: Session, user_id: str, concept_key: str) -> list[RepairItem]:
    return list(
        db.scalars(
            select(RepairItem).where(
                RepairItem.user_id == user_id,
                RepairItem.concept_key == concept_key,
                RepairItem.status.in_(("open", "escalated")),
            )
        )
    )


def _apply_checkpoint_task_consequence(
    db: Session,
    *,
    user_id: str,
    payload: CheckpointRequest,
    correct: bool,
    repair_status: str,
    now: datetime,
) -> bool:
    """Complete only an exact task whose required evidence this answer satisfies."""
    if not payload.daily_task_id:
        return False
    task = db.scalar(
        select(DailyTask).where(
            DailyTask.id == payload.daily_task_id,
            DailyTask.user_id == user_id,
        )
    )
    if task is None:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Study task not found.")
    if task.mission_id and task.mission_id != payload.mission_id:
        raise HTTPException(
            status_code=status.HTTP_409_CONFLICT,
            detail={"code": "TASK_EVIDENCE_MISMATCH", "message": "This answer belongs to a different lesson task."},
        )
    evidence_completes = (
        correct
        and not payload.hint_used
        and payload.attempt_index == 1
        and (
            (task.task_type == "mistake_repair" and repair_status == "resolved")
            or (task.task_type == "revision" and payload.kind in {"revision_recall", "transfer_check", "board_answer"})
            or (task.task_type == "board_answer_practice" and payload.kind == "board_answer")
        )
    )
    if not evidence_completes or task.status == "completed":
        return False
    task.status = "completed"
    task.completed_at = now
    task.task_metadata = {
        **dict(task.task_metadata or {}),
        "completion_evidence": {
            "kind": "checkpoint_event",
            "client_event_id": payload.client_event_id,
            "question_id": payload.question_id,
        },
    }
    refresh_daily_plan_totals(db, plan_id=task.daily_plan_id, user_id=user_id)
    add_learning_event(
        db,
        user_id=user_id,
        event_type="TASK_COMPLETED",
        entity_type="daily_task",
        entity_id=task.id,
        idempotency_key=f"task-completed:checkpoint:{payload.client_event_id}:{task.id}",
        subject_id=task.subject_id,
        chapter_id=task.chapter_id,
        topic_key=payload.concept_key,
        event_data={"daily_plan_id": task.daily_plan_id, "evidence_type": "checkpoint_event"},
    )
    return True


def record_checkpoint(
    db: Session,
    *,
    user_id: str,
    payload: CheckpointRequest,
    now: datetime | None = None,
    planning_now: datetime | None = None,
    trusted_correct: bool | None = None,
) -> CheckpointConsequenceResponse:
    # Idempotency: a replayed client_event_id returns the stored consequence.
    existing = db.scalar(
        select(CheckpointEvent).where(
            CheckpointEvent.user_id == user_id,
            CheckpointEvent.client_event_id == payload.client_event_id,
        )
    )
    if existing is not None:
        source_id = dict(existing.payload or {}).get("source_question_id")
        if (existing.concept_key != payload.concept_key
                or payload.question_id not in {existing.question_id, source_id}
                or dict(existing.payload or {}).get("student_answer") != payload.student_answer):
            raise HTTPException(409, detail={"code": "CHECKPOINT_REPLAY_MISMATCH",
                                            "message": "This saved attempt belongs to a different answer. Start a new attempt."})
        stored = dict(existing.consequence or {})
        stored["replayed"] = True
        return CheckpointConsequenceResponse.model_validate(stored)

    now = aware(now) or _now()
    plan_clock = aware(planning_now) or now
    source_question_id = payload.question_id
    authored_source_id = source_question_id
    if trusted_correct is None:
        payload, correct, authored_source_id = evaluate_checkpoint(payload)
    else:
        # Private Python-only path for the server-owned written rubric. The
        # HTTP schema forbids this field; browser flags can never reach it.
        correct = trusted_correct
    # Canonicalization must not make a pre-release raw question ID look fresh.
    # Treat durable attempts conservatively, including previously client-graded
    # records: they can establish familiarity, never manufacture new success.
    question_aliases = {payload.question_id, source_question_id, authored_source_id}
    if trusted_correct is None:
        question_aliases.add(payload.question_id.rsplit(":", 1)[-1])
    previous_attempts = list(db.scalars(select(CheckpointEvent).where(
        CheckpointEvent.user_id == user_id,
        CheckpointEvent.concept_key == payload.concept_key,
        CheckpointEvent.question_id.in_(question_aliases),
    )))

    error_category: str | None = None
    if not correct:
        error_category = classify_error(
            question_type=payload.question_type,
            student_answer=payload.student_answer,
            correct_answer=payload.correct_answer,
            expected_keywords=payload.expected_keywords,
        )

    profile = db.scalar(select(StudentProfileState).where(StudentProfileState.user_id == user_id))
    exam_date = profile.exam_date if profile is not None else None
    chapter = _owned_chapter_for_checkpoint(db, user_id=user_id, payload=payload)
    resolved_chapter_id = (chapter.id if chapter else None) or payload.chapter_id
    resolved_subject_id = (chapter.subject_id if chapter else None) or payload.subject_id

    # --- Repair lifecycle -------------------------------------------------
    open_repairs = _open_repairs_for_concept(db, user_id, payload.concept_key)
    independent_repair_check = False
    if correct and open_repairs and not payload.hint_used and payload.attempt_index == 1:
        # A correction to an already-seen question is practice, not proof of
        # transfer. Use persisted attempts rather than the bounded mastery log
        # so restarting the class cannot disguise a retry as a fresh check.
        independent_repair_check = not previous_attempts
    if correct and open_repairs:
        logger.debug(
            "[FIX:independent-repair] checked recovery evidence",
            extra={"assisted": payload.hint_used or payload.attempt_index != 1,
                   "independent_repair_check": independent_repair_check},
        )
    repair_status: str = "none"
    repair_item: RepairItem | None = None

    if not correct and error_category is not None:
        matching = next((item for item in open_repairs if item.error_category == error_category), None)
        node = concept_graph.concept_for(payload.concept_key)
        importance = node.exam_importance if node else 0.7
        if matching is not None:
            matching.failed_attempts += 1
            matching.support_level = min(3, matching.support_level + 1)
            matching.priority = round(min(2.5, matching.priority + 0.2), 2)
            matching.status = "escalated"
            matching.chapter_id = matching.chapter_id or resolved_chapter_id
            matching.subject_id = matching.subject_id or resolved_subject_id
            matching.mission_id = matching.mission_id or payload.mission_id
            matching.evidence = {
                **(matching.evidence or {}),
                "last_failed_question_id": payload.question_id,
                "last_failed_at": now.isoformat(),
            }
            repair_item = matching
            repair_status = "escalated"
        else:
            copy = ERROR_CATEGORY_COPY.get(error_category, ERROR_CATEGORY_COPY["concept_misunderstanding"])
            repair_item = RepairItem(
                user_id=user_id,
                subject_id=resolved_subject_id,
                chapter_id=resolved_chapter_id,
                concept_key=payload.concept_key,
                concept_label=payload.concept_label,
                mission_id=payload.mission_id,
                error_category=error_category,
                diagnosis=copy["diagnosis"],
                recommended_activity=copy["activity"],
                activity_prompt=copy["activity_prompt"],
                priority=round(1.6 + importance * 0.4, 2),
                estimated_minutes=REPAIR_TASK_MINUTES,
                status="open",
                support_level=1,
                failed_attempts=1,
                source_kind="checkpoint",
                source_ref=payload.question_id,
                evidence={
                    "question_id": payload.question_id,
                    "prompt": payload.prompt[:500],
                    "student_answer": payload.student_answer[:500],
                    "at": now.isoformat(),
                },
            )
            db.add(repair_item)
            repair_status = "created"
    elif independent_repair_check:
        # Only a fresh, unhinted first attempt closes the concept's open
        # repairs. Supported practice is still recorded by mastery below.
        for item in open_repairs:
            item.status = "resolved"
            item.retry_result = "recovered"
            item.resolved_at = now
        repair_item = open_repairs[0]
        repair_status = "resolved"

    # --- Mastery evidence ---------------------------------------------------
    mastery_row = db.scalar(
        select(TopicMastery)
        .where(TopicMastery.user_id == user_id, TopicMastery.topic_key == payload.concept_key)
        .with_for_update()
    )
    snapshot = MasterySnapshot(
        score=(mastery_row.score if mastery_row else None) or 0.0,
        confidence=(mastery_row.confidence if mastery_row else None) or 0.0,
        attempts_count=(mastery_row.attempts_count if mastery_row else None) or 0,
        state=(mastery_row.last_result if mastery_row and mastery_row.last_result else "not_started"),
        next_review_at=mastery_row.next_review_at if mastery_row else None,
        evidence=dict(mastery_row.evidence or {}) if mastery_row else {},
    )
    if any(attempt.correct for attempt in previous_attempts):
        seen = list(snapshot.evidence.get("seen_question_correct") or [])
        if payload.question_id not in seen:
            snapshot.evidence["seen_question_correct"] = [*seen, payload.question_id]
    # Capture the BEFORE base/overlay for mastery transition instrumentation
    # (M7). An overlay clearing is logged separately and must never count as
    # a base mastery gain.
    before_base = base_mastery_state(snapshot.state or "not_started")
    before_overlay = overlay_for(snapshot.state or "not_started")
    # After the lifecycle above, the concept still has an open repair when one
    # was just created/escalated, or pre-existing opens were not resolved here.
    has_open_repair_after = repair_status in {"created", "escalated"} or (
        repair_status != "resolved" and bool(open_repairs)
    )
    update = apply_evidence(
        snapshot,
        EvidenceEvent(
            kind=payload.kind,
            correct=correct,
            at=now,
            question_id=payload.question_id,
            hint_used=payload.hint_used,
            attempt_index=payload.attempt_index,
            error_category=error_category,
            time_spent_seconds=payload.time_spent_seconds,
            source="guided_class",
        ),
        exam_date=exam_date,
        has_open_repair=has_open_repair_after,
    )

    if mastery_row is None:
        mastery_row = TopicMastery(
            user_id=user_id,
            subject_id=resolved_subject_id,
            chapter_id=resolved_chapter_id,
            topic_key=payload.concept_key,
            topic_label=payload.concept_label,
        )
        db.add(mastery_row)
    mastery_row.subject_id = resolved_subject_id or mastery_row.subject_id
    mastery_row.chapter_id = resolved_chapter_id or mastery_row.chapter_id
    mastery_row.topic_label = payload.concept_label
    mastery_row.score = snapshot.score
    mastery_row.confidence = snapshot.confidence
    mastery_row.attempts_count = snapshot.attempts_count
    mastery_row.last_result = snapshot.state
    mastery_row.next_review_at = snapshot.next_review_at
    mastery_row.evidence = snapshot.evidence
    _sync_revision_item_from_checkpoint(
        db,
        user_id=user_id,
        payload=payload,
        chapter=chapter,
        next_review_at=snapshot.next_review_at,
        now=now,
        correct=correct,
    )

    if repair_item is not None and repair_status == "resolved":
        repair_item.mastery_recovered = snapshot.score >= 60.0

    # --- Message the student can trust ---------------------------------------
    if correct and (payload.hint_used or payload.attempt_index != 1):
        message = "Correct with support. Try a fresh question without hints to check independent understanding."
    elif correct and open_repairs and not independent_repair_check:
        message = "Correct on a familiar question. The repair stays in your plan until a fresh independent check."
    elif correct and repair_status == "resolved":
        message = "Repair recovered — this concept is back on track and your plan will drop the repair task."
    elif correct:
        message = "Correct. This strengthens the concept's mastery and pushes its next revision further out."
    elif repair_status == "escalated":
        message = "Still stuck on the same kind of mistake — DocDoe raised the support level for this repair."
    else:
        message = "Not yet. DocDoe recorded the exact mistake and added a short repair to today's plan."

    mastery_out = CheckpointMasteryOut(
        before_score=update.before_score,
        after_score=update.after_score,
        before_state=update.before_state,
        after_state=update.after_state,
        confidence=update.confidence,
        consecutive_success=update.consecutive_success,
        next_review_at=update.next_review_at,
    )

    db.flush()  # so repair_item.id exists for the stored consequence

    consequence = CheckpointConsequenceResponse(
        checkpoint_id="pending",
        correct=correct,
        replayed=False,
        error_category=error_category,
        diagnosis=repair_item.diagnosis if repair_item is not None and not correct else None,
        recommended_activity=repair_item.recommended_activity if repair_item is not None and not correct else None,
        activity_prompt=repair_item.activity_prompt if repair_item is not None and not correct else None,
        repair_status=repair_status,  # type: ignore[arg-type]
        repair_item=repair_item_out(repair_item) if repair_item is not None else None,
        mastery=mastery_out,
        message=message,
    )

    event = CheckpointEvent(
        user_id=user_id,
        client_event_id=payload.client_event_id,
        concept_key=payload.concept_key,
        question_id=payload.question_id,
        attempt_index=payload.attempt_index,
        kind=payload.kind,
        correct=correct,
        error_category=error_category,
        hint_used=payload.hint_used,
        payload={
            "question_type": payload.question_type,
            "prompt": payload.prompt[:1000],
            "student_answer": payload.student_answer,
            "correct_answer": payload.correct_answer[:1000],
            "client_correct": payload.client_correct,
            "source_question_id": source_question_id,
            "evaluation_version": EVALUATION_VERSION if trusted_correct is None else "server-written-rubric",
            "expected_keywords": payload.expected_keywords,
            "time_spent_seconds": payload.time_spent_seconds,
            "difficulty": payload.difficulty,
            "skill_tags": payload.skill_tags,
            "error_reflection": (
                payload.error_reflection.model_dump()
                if payload.error_reflection is not None and not correct
                else None
            ),
        },
        consequence={},
    )
    db.add(event)
    try:
        db.flush()
    except IntegrityError:
        # A parallel tab won the unique-constraint race; return its consequence.
        db.rollback()
        winner = db.scalar(
            select(CheckpointEvent).where(
                CheckpointEvent.user_id == user_id,
                CheckpointEvent.client_event_id == payload.client_event_id,
            )
        )
        if winner is None:
            raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="Checkpoint conflict; retry.")
        if (winner.concept_key != payload.concept_key
                or winner.question_id not in question_aliases
                or dict(winner.payload or {}).get("student_answer") != payload.student_answer):
            raise HTTPException(409, detail={"code": "CHECKPOINT_REPLAY_MISMATCH",
                                            "message": "This saved attempt belongs to a different answer. Start a new attempt."})
        stored = dict(winner.consequence or {})
        stored["replayed"] = True
        return CheckpointConsequenceResponse.model_validate(stored)

    consequence.checkpoint_id = event.id
    event.consequence = consequence.model_dump(mode="json")
    db.add(
        UsageEvent(
            user_id=user_id,
            event_type="checkpoint_answered",
            resource_type="checkpoint_event",
            event_data={"checkpoint_id": event.id, "concept_key": payload.concept_key, "correct": correct},
        )
    )
    # --- Instrument A: reliable event vocabulary -----------------------------
    # Every checkpoint emits a CHECKPOINT_RECORDED with first-attempt + hint
    # metadata. Repair lifecycle and REPAIR_TASK_OPENED are emitted separately
    # so analysts can reconstruct the full student flow without ambiguity.
    attempt_index = payload.attempt_index
    first_attempt_correct = attempt_index == 1 and correct and not payload.hint_used
    add_learning_event(
        db,
        user_id=user_id,
        event_type=EVENT_CHECKPOINT_RECORDED,
        entity_type="checkpoint_event",
        entity_id=str(event.id),
        idempotency_key=f"checkpoint:{payload.client_event_id}",
        subject_id=payload.subject_id,
        chapter_id=payload.chapter_id,
        topic_key=payload.concept_key,
        event_data={
            "question_id": payload.question_id,
            "correct": correct,
            "attempt_index": attempt_index,
            "first_attempt_correct": first_attempt_correct,
            "hint_used": payload.hint_used,
            "error_category": error_category,
            "repair_status": repair_status,
            "streak_credited": update.streak_credited,
            "before_score": update.before_score,
            "after_score": update.after_score,
            "before_state": update.before_state,
            "after_state": update.after_state,
            "difficulty": payload.difficulty,
            "skill_tags": payload.skill_tags,
            "reported_error_type": (
                payload.error_reflection.error_type
                if payload.error_reflection is not None and not correct
                else None
            ),
        },
    )
    if payload.hint_used:
        add_learning_event(
            db,
            user_id=user_id,
            event_type=EVENT_HINT_USED,
            entity_type="checkpoint_event",
            entity_id=str(event.id),
            idempotency_key=f"hint:{payload.client_event_id}",
            subject_id=payload.subject_id,
            chapter_id=payload.chapter_id,
            topic_key=payload.concept_key,
            event_data={
                "question_id": payload.question_id,
                "attempt_index": attempt_index,
                "concept_key": payload.concept_key,
            },
        )
    # --- Repair lifecycle events ---------------------------------------------
    if repair_status == "created":
        add_learning_event(
            db,
            user_id=user_id,
            event_type=EVENT_REPAIR_CREATED,
            entity_type="checkpoint_event",
            entity_id=str(event.id),
            idempotency_key=f"repair-created:{payload.client_event_id}",
            subject_id=payload.subject_id,
            chapter_id=payload.chapter_id,
            topic_key=payload.concept_key,
            event_data={
                "question_id": payload.question_id,
                "error_category": error_category,
                "concept_key": payload.concept_key,
            },
        )
    elif repair_status == "escalated":
        add_learning_event(
            db,
            user_id=user_id,
            event_type=EVENT_REPAIR_ESCALATED,
            entity_type="checkpoint_event",
            entity_id=str(event.id),
            idempotency_key=f"repair-escalated:{payload.client_event_id}",
            subject_id=payload.subject_id,
            chapter_id=payload.chapter_id,
            topic_key=payload.concept_key,
            event_data={
                "question_id": payload.question_id,
                "error_category": error_category,
                "concept_key": payload.concept_key,
            },
        )
    elif repair_status == "resolved":
        add_learning_event(
            db,
            user_id=user_id,
            event_type=EVENT_REPAIR_RESOLVED,
            entity_type="checkpoint_event",
            entity_id=str(event.id),
            idempotency_key=f"repair-resolved:{payload.client_event_id}",
            subject_id=payload.subject_id,
            chapter_id=payload.chapter_id,
            topic_key=payload.concept_key,
            event_data={
                "question_id": payload.question_id,
                "error_category": error_category,
                "concept_key": payload.concept_key,
                "mastery_recovered": repair_item.mastery_recovered if repair_item else None,
            },
        )
    # --- REPAIR_TASK_OPENED: distinct from REPAIR_CREATED ---------------------
    # Item 8 of the brief: a student opening a repair task (the *act of
    # starting* the repair) is distinct from the repair being *created*.
    # We emit this when a checkpoint is recorded against a mistake_repair task
    # and the student has not yet resolved it (i.e., "never attempted" vs
    # "attempted and failed" must not be conflated).
    if payload.daily_task_id:
        _task_for_event = db.scalar(
            select(DailyTask).where(
                DailyTask.id == payload.daily_task_id,
                DailyTask.user_id == user_id,
            )
        )
        if _task_for_event is not None and _task_for_event.task_type == "mistake_repair":
            _repair_attempted = _task_for_event.status not in ("not_started", "pending")
            add_learning_event(
                db,
                user_id=user_id,
                event_type=EVENT_REPAIR_TASK_OPENED,
                entity_type="daily_task",
                entity_id=str(_task_for_event.id),
                idempotency_key=f"repair-opened:{payload.client_event_id}:{_task_for_event.id}",
                subject_id=payload.subject_id,
                chapter_id=payload.chapter_id,
                topic_key=payload.concept_key,
                event_data={
                    "concept_key": payload.concept_key,
                    "repair_status": repair_status,
                    "previously_attempted": _repair_attempted,
                    "question_id": payload.question_id,
                },
            )
    # --- MASTERY_STATE_CHANGED (base only, never overlays) -------------------
    # M7 measures base-state movement only. Overlay-only changes are logged
    # separately and must never count as mastery gains.
    record_mastery_transition(
        db,
        user_id=user_id,
        topic_key=payload.concept_key,
        before_base_state=before_base,
        after_base_state=update.after_base_state,
        before_overlay=before_overlay,
        after_overlay=update.after_overlay,
        trigger_entity_type="checkpoint_event",
        trigger_entity_id=str(event.id),
        subject_id=payload.subject_id,
        chapter_id=payload.chapter_id,
        algorithm_version=f"mastery_engine_v{MASTERY_ENGINE_VERSION}",
        evidence_count=snapshot.attempts_count,
    )
    _apply_checkpoint_task_consequence(
        db,
        user_id=user_id,
        payload=payload,
        correct=correct,
        repair_status=repair_status,
        now=now,
    )
    db.commit()
    try:
        replan_today(
            db,
            user_id=user_id,
            idempotency_key=f"checkpoint:{payload.client_event_id}",
            reason="meaningful_evidence",
            now=plan_clock,
        )
    except HTTPException as exc:
        if exc.status_code != status.HTTP_409_CONFLICT:
            raise
        logger.info("today_plan_not_replanned user_id=%s reason=study_plan_required", user_id)
    logger.info(
        "learning.checkpoint_applied",
        extra={
            "user_ref": user_id[-8:],
            "checkpoint_ref": event.id,
            "topic_key": payload.concept_key,
            "correct": correct,
            "error_category": error_category,
            "reported_error_type": (
                payload.error_reflection.error_type
                if payload.error_reflection is not None and not correct
                else None
            ),
            "repair_status": repair_status,
            "mastery_before": update.before_score,
            "mastery_after": update.after_score,
            "next_review_at": (
                update.next_review_at.isoformat() if update.next_review_at else None
            ),
        },
    )
    return consequence


# ---------------------------------------------------------------------------
# Compatibility
# ---------------------------------------------------------------------------


def plan_today(db: Session, *, user_id: str) -> PlanTodayResponse:
    """Compatibility wrapper for older imports.
    The persisted Learning Engine in today_plan_service is the sole
    executable planning authority.
    """
    from app.services.today_plan_service import ensure_today_plan

    return ensure_today_plan(db, user_id=user_id)