"""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)