app/services/learning_state_service.py

from __future__ import annotations

from datetime import date, datetime, time, timedelta, timezone
from typing import Iterable

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

from app.core.http_status import HTTP_422_UNPROCESSABLE_CONTENT
from app.models.chat_session import ChatMessageRecord, ChatSession
from app.models.class_session_progress import ClassSessionProgress
from app.models.learn_anything_roadmap import LearnAnythingRoadmap
from app.models.learning_state import (
    Chapter,
    DailyPlan,
    DailyTask,
    GeneratedResource,
    RevisionItem,
    LessonProgress,
    QuizAttempt,
    RepairItem,
    StudyFollowUp,
    StudentProfileState,
    StudyPlan,
    StudySession,
    Subject,
    TopicMastery,
    UsageEvent,
)
from app.models.study_profile import StudyProfile
from app.services.adaptive_engine import repair_item_out
from app.services.academic_state import build_academic_projection, current_mastery_state
from app.services.learning_events import add_learning_event
from app.services.today_plan_service import daily_plan_out, refresh_daily_plan_totals, replan_today
from app.schemas.learning_state import (
    AssessmentConsequenceResponse,
    AssessmentResultRequest,
    LearningChapterOut,
    LearningLessonProgressOut,
    LearningOnboardingRequest,
    LearningOnboardingResponse,
    LessonProgressRequest,
    LessonProgressResponse,
    LearningProfileOut,
    LearningQuizAttemptOut,
    LearningResourceOut,
    LearningRevisionItemOut,
    LearningStateSummary,
    LearningSubjectOut,
    LearningTaskOut,
    PlanAdjustmentRequest,
    PlanAdjustmentResponse,
    RevisionItemSaveRequest,
    RevisionItemSaveResponse,
    RevisionResultRequest,
    RevisionResultResponse,
    StudyFollowUpCreateRequest,
    StudyFollowUpOut,
    TaskStatusResponse,
    TopicMasteryOut,
)


_SOUND_WAVES_STEPS = (
    ("Oscillation, amplitude, period and frequency", "lesson", 35, "M1"),
    ("Natural frequency, forced vibration and resonance", "lesson", 35, "M2"),
    ("Wave motion and types of waves", "lesson", 35, "M3"),
    ("Frequency, wavelength and wave speed", "practice", 35, "M4"),
    ("Numericals using v = fλ", "practice", 40, "M5"),
    ("Reflection, echo and reverberation", "revision", 35, "M6"),
    ("Limits of audibility and ultrasonic uses", "revision", 30, "M7"),
)


def _is_sound_waves_ready(payload: LearningOnboardingRequest) -> bool:
    return (
        any(subject.casefold() == "physics" for subject in payload.subjects)
        and "kerala" in payload.board.casefold()
        and ("10" in payload.class_level or "sslc" in payload.class_level.casefold())
    )


def _scheduled_datetime(day: date, preferred_time: str | None) -> datetime:
    label = (preferred_time or "evening").casefold()
    hour = 7 if "morning" in label else 20 if "night" in label else 18
    return datetime.combine(day, time(hour=hour), tzinfo=timezone.utc)


def _task_out(task: DailyTask, subjects: dict[str, str], chapters: dict[str, str]) -> LearningTaskOut:
    metadata = task.task_metadata or {}
    return LearningTaskOut(
        id=task.id,
        study_plan_id=task.study_plan_id,
        daily_plan_id=task.daily_plan_id,
        logical_key=task.logical_key,
        subject_id=task.subject_id,
        chapter_id=task.chapter_id,
        subject=subjects.get(task.subject_id or ""),
        chapter=chapters.get(task.chapter_id or ""),
        task_type=task.task_type,
        title=task.title,
        status=task.status,
        scheduled_for=task.scheduled_for,
        duration_minutes=task.duration_minutes,
        priority=task.priority,
        href=task.href,
        mission_id=task.mission_id,
        rescheduled_from_id=task.rescheduled_from_id,
        completed_at=task.completed_at,
        reason=metadata.get("reason"),
        source=metadata.get("source"),
        repair_item_id=metadata.get("repair_item_id"),
        revision_concept_key=metadata.get("revision_concept_key"),
        reason_codes=list(metadata.get("reason_codes") or []),
        reason_facts=dict(metadata.get("reason_facts") or {}),
        target=dict(metadata.get("target") or {}),
    )


def _task_maps(db: Session, user_id: str) -> tuple[dict[str, str], dict[str, str]]:
    subjects = {item.id: item.name for item in db.scalars(select(Subject).where(Subject.user_id == user_id)).all()}
    chapters = {item.id: item.title for item in db.scalars(select(Chapter).where(Chapter.user_id == user_id)).all()}
    return subjects, chapters


def _mastery_out(item: TopicMastery, *, resolved_state: str | None = None) -> TopicMasteryOut:
    """Expose the mastery engine's evidence-derived fields alongside the row."""
    evidence = item.evidence or {}
    last_correct = evidence.get("last_correct_at")
    last_wrong = evidence.get("last_wrong_at")
    last_evidence: datetime | None = None
    for raw in (last_correct, last_wrong):
        if isinstance(raw, str):
            try:
                candidate = datetime.fromisoformat(raw)
            except ValueError:
                continue
            if last_evidence is None or candidate > last_evidence:
                last_evidence = candidate
    return TopicMasteryOut(
        id=item.id,
        subject_id=item.subject_id,
        chapter_id=item.chapter_id,
        topic_key=item.topic_key,
        topic_label=item.topic_label,
        score=item.score,
        confidence=item.confidence,
        attempts_count=item.attempts_count,
        last_result=item.last_result,
        next_review_at=item.next_review_at,
        state=resolved_state or item.last_result or "not_started",
        consecutive_success=int(evidence.get("consecutive_success", 0) or 0),
        error_categories={
            str(key): int(value)
            for key, value in (evidence.get("error_categories", {}) or {}).items()
            if isinstance(value, (int, float))
        },
        last_evidence_at=last_evidence,
    )


def _replan_if_setup_exists(
    db: Session,
    *,
    user_id: str,
    idempotency_key: str,
    reason: str,
    now: datetime | None = None,
):
    """Replan after a meaningful event without breaking pre-setup saves."""

    try:
        return replan_today(
            db,
            user_id=user_id,
            idempotency_key=idempotency_key,
            reason=reason,
            now=now,
        )
    except HTTPException as exc:
        detail = exc.detail if isinstance(exc.detail, dict) else {}
        if exc.status_code == status.HTTP_409_CONFLICT and detail.get("code") == "STUDY_PLAN_REQUIRED":
            return None
        raise


def _revision_item_out(
    item: RevisionItem,
    subjects: dict[str, str],
    chapters: dict[str, str],
) -> LearningRevisionItemOut:
    return LearningRevisionItemOut(
        id=item.id,
        client_item_id=item.client_item_id,
        subject_id=item.subject_id,
        chapter_id=item.chapter_id,
        subject=subjects.get(item.subject_id or ""),
        chapter=chapters.get(item.chapter_id or ""),
        mission_id=item.mission_id,
        topic_key=item.topic_key,
        title=item.title,
        source_kind=item.source_kind,
        source_ref=item.source_ref,
        status=item.status,  # type: ignore[arg-type]
        due_at=item.due_at,
        completed_at=item.completed_at,
        created_at=item.created_at,
    )


def _mirror_onboarding_study_profile(
    db: Session,
    *,
    user_id: str,
    payload: LearningOnboardingRequest,
    available_days: int,
) -> StudyProfile:
    """Keep the legacy tutor context in the onboarding transaction.

    StudyChat, source routing and the onboarding gate still read
    ``study_profiles`` while the adaptive planner owns ``student_profiles``.
    Writing both rows before the same commit prevents a completed profile from
    existing without the plan and first task that completion promises.
    """

    profile = db.scalar(
        select(StudyProfile)
        .where(StudyProfile.user_id == user_id)
        .with_for_update()
    )
    if profile is None:
        profile = StudyProfile(user_id=user_id)

    preferences = dict(payload.preferences or {})
    language = preferences.get("language")
    learning_style = preferences.get("learning_style")
    focus_areas = preferences.get("focus_areas")
    daily_time = preferences.get("daily_time")
    preferred_time = preferences.get("preferred_time")
    if not isinstance(daily_time, str) or not daily_time.strip():
        daily_time = (
            f"{payload.daily_minutes // 60} hours"
            if payload.daily_minutes >= 120 and payload.daily_minutes % 60 == 0
            else f"{payload.daily_minutes} minutes"
        )
    if not isinstance(preferred_time, str) or not preferred_time.strip():
        preferred_time = payload.preferred_time

    extra = dict(profile.extra or {})
    extra.update(
        {
            "schema_version": 2,
            "subjects": list(payload.subjects),
            "daily_time": daily_time,
            "daily_minutes": payload.daily_minutes,
            "preferred_time": preferred_time,
            "focus_areas": focus_areas if isinstance(focus_areas, list) else [],
            "language_preference": language if isinstance(language, str) else None,
            "learning_style": learning_style if isinstance(learning_style, str) else None,
            "available_study_days": available_days,
            "first_plan_days": min(7, available_days),
            "setup_completed_at": datetime.now(timezone.utc).isoformat(),
        }
    )

    profile.board = payload.board
    profile.grade = payload.class_level
    profile.subject = payload.subjects[0]
    profile.goal = payload.goal
    profile.time_left = payload.exam_date.isoformat() if payload.exam_date else None
    profile.language_preference = language if isinstance(language, str) else None
    profile.source_mode = "onboarding"
    profile.onboarding_completed = 1
    profile.extra = extra
    db.add(profile)
    return profile


def create_onboarding_plan(
    db: Session,
    *,
    user_id: str,
    payload: LearningOnboardingRequest,
) -> LearningOnboardingResponse:
    today = date.today()
    # Exam date is optional. When the student does not know it yet, plan a
    # steady default horizon rather than trapping them at onboarding.
    DEFAULT_UNKNOWN_HORIZON_DAYS = 90
    if payload.exam_date is None:
        available_days = DEFAULT_UNKNOWN_HORIZON_DAYS
    else:
        available_days = (payload.exam_date - today).days
        if available_days <= 0:
            raise HTTPException(
                status_code=HTTP_422_UNPROCESSABLE_CONTENT,
                detail={"code": "EXAM_DATE_NOT_FUTURE", "message": "Choose an exam date after today."},
            )

    profile = db.scalar(select(StudentProfileState).where(StudentProfileState.user_id == user_id).with_for_update())
    if profile is None:
        profile = StudentProfileState(
            user_id=user_id,
            class_level=payload.class_level,
            board=payload.board,
            exam_date=payload.exam_date,
            goal=payload.goal,
            daily_minutes=payload.daily_minutes,
            preferred_time=payload.preferred_time,
            available_study_days=available_days,
            preferences=payload.preferences,
        )
        db.add(profile)
        db.flush()
    else:
        profile.class_level = payload.class_level
        profile.board = payload.board
        profile.exam_date = payload.exam_date
        profile.goal = payload.goal
        profile.daily_minutes = payload.daily_minutes
        profile.preferred_time = payload.preferred_time
        profile.available_study_days = available_days
        profile.preferences = payload.preferences
        profile.onboarding_completed = True

    existing_subjects = {
        item.name.casefold(): item
        for item in db.scalars(select(Subject).where(Subject.user_id == user_id)).all()
    }
    selected: list[Subject] = []
    selected_keys = {value.casefold() for value in payload.subjects}
    for item in existing_subjects.values():
        item.status = "active" if item.name.casefold() in selected_keys else "inactive"
    for priority, name in enumerate(payload.subjects):
        subject = existing_subjects.get(name.casefold())
        if subject is None:
            subject = Subject(
                user_id=user_id,
                name=name,
                board=payload.board,
                class_level=payload.class_level,
                priority=priority,
                status="active",
            )
            db.add(subject)
            db.flush()
        else:
            subject.board = payload.board
            subject.class_level = payload.class_level
            subject.priority = priority
            subject.status = "active"
        selected.append(subject)

    sound_chapter: Chapter | None = None
    physics = next((item for item in selected if item.name.casefold() == "physics"), None)
    if physics is not None and _is_sound_waves_ready(payload):
        sound_chapter = db.scalar(
            select(Chapter).where(
                Chapter.user_id == user_id,
                Chapter.subject_id == physics.id,
                Chapter.catalog_id == "phy-p1-c1",
            )
        )
        if sound_chapter is None:
            sound_chapter = Chapter(
                user_id=user_id,
                subject_id=physics.id,
                catalog_id="phy-p1-c1",
                title="Sound Waves",
                order_index=1,
                importance=1.0,
                estimated_minutes=240,
                status="available",
                curated=True,
            )
            db.add(sound_chapter)
            db.flush()

    active_plans = db.scalars(
        select(StudyPlan).where(StudyPlan.user_id == user_id, StudyPlan.status == "active")
    ).all()
    for previous in active_plans:
        previous.status = "superseded"

    plan = StudyPlan(
        user_id=user_id,
        name="First adaptive study week",
        status="active",
        start_date=today,
        end_date=(
            today + timedelta(days=6)
            if payload.exam_date is None
            else min(payload.exam_date, today + timedelta(days=6))
        ),
        daily_minutes=payload.daily_minutes,
        algorithm_version="adaptive-v1",
        planning_factors={
            "syllabus_remaining": "verified_catalog_only",
            "days_until_exam": available_days,
            "daily_available_time": payload.daily_minutes,
            "topic_importance": "enabled",
            "mastery": "unknown_until_assessed",
            "recent_mistakes": [],
            "unfinished_tasks": [],
            "revision_spacing": [1, 3, 7],
        },
    )
    db.add(plan)
    db.flush()

    tasks: list[DailyTask] = []
    physics_step = 0
    for day_offset in range(min(7, available_days)):
        subject = selected[day_offset % len(selected)]
        scheduled_day = today + timedelta(days=day_offset)
        if sound_chapter is not None and subject.id == physics.id:
            title, task_type, duration, mission_id = _SOUND_WAVES_STEPS[min(physics_step, len(_SOUND_WAVES_STEPS) - 1)]
            physics_step += 1
            task = DailyTask(
                user_id=user_id,
                study_plan_id=plan.id,
                subject_id=subject.id,
                chapter_id=sound_chapter.id,
                task_type=task_type,
                title=f"Sound Waves: {title}",
                status="pending",
                scheduled_for=_scheduled_datetime(scheduled_day, payload.preferred_time),
                duration_minutes=min(payload.daily_minutes, duration),
                priority=1.0,
                href="/tuition",
                mission_id=mission_id,
                task_metadata={"source": "curated", "chapter_catalog_id": "phy-p1-c1"},
            )
        else:
            task = DailyTask(
                user_id=user_id,
                study_plan_id=plan.id,
                subject_id=subject.id,
                task_type="upload",
                title=f"Choose your {subject.name} material before DocDoe plans this lesson",
                status="pending",
                scheduled_for=_scheduled_datetime(scheduled_day, payload.preferred_time),
                duration_minutes=min(payload.daily_minutes, 15),
                priority=0.7,
                href=f"/study-chat?prompt=Help%20me%20choose%20my%20{subject.name.replace(' ', '%20')}%20material",
                task_metadata={"source": "student_material_required"},
            )
        db.add(task)
        tasks.append(task)
    db.flush()

    first_task = tasks[0]
    profile.current_subject_id = first_task.subject_id
    profile.current_chapter_id = first_task.chapter_id
    profile.current_mission_id = first_task.mission_id
    _mirror_onboarding_study_profile(
        db,
        user_id=user_id,
        payload=payload,
        available_days=available_days,
    )
    db.add(
        UsageEvent(
            user_id=user_id,
            event_type="onboarding_completed",
            resource_type="study_plan",
            event_data={"plan_id": plan.id, "available_study_days": available_days},
        )
    )
    db.commit()

    # Onboarding is the explicit setup transition that creates the first
    # persisted Today Plan. Home can now remain a read-only consumer.
    from app.services.today_plan_service import ensure_today_plan

    today_plan_response = ensure_today_plan(db, user_id=user_id)
    persisted_today_plan = db.scalar(
        select(DailyPlan).where(DailyPlan.id == today_plan_response.plan_id)
    )

    subjects_map, chapters_map = _task_maps(db, user_id)
    return LearningOnboardingResponse(
        profile_id=profile.id,
        plan_id=plan.id,
        available_study_days=available_days,
        first_task_id=first_task.id,
        first_task_href=first_task.href or "/home",
        tasks=[_task_out(task, subjects_map, chapters_map) for task in tasks],
        today_plan=(daily_plan_out(db, persisted_today_plan) if persisted_today_plan else None),
    )


def read_learning_state(db: Session, *, user_id: str) -> LearningStateSummary:
    profile = db.scalar(select(StudentProfileState).where(StudentProfileState.user_id == user_id))
    subjects = db.scalars(select(Subject).where(Subject.user_id == user_id).order_by(Subject.priority, Subject.name)).all()
    chapters = db.scalars(select(Chapter).where(Chapter.user_id == user_id).order_by(Chapter.order_index)).all()
    active_plan = db.scalar(
        select(StudyPlan).where(
            StudyPlan.user_id == user_id,
            StudyPlan.status == "active",
        ).order_by(StudyPlan.created_at.desc())
    )
    tasks = (
        db.scalars(
            select(DailyTask).where(
                DailyTask.user_id == user_id,
                DailyTask.study_plan_id == active_plan.id,
            ).order_by(DailyTask.scheduled_for)
        ).all()
        if active_plan is not None
        else []
    )
    mastery = db.scalars(select(TopicMastery).where(TopicMastery.user_id == user_id).order_by(TopicMastery.score)).all()
    repair_items = db.scalars(
        select(RepairItem)
        .where(RepairItem.user_id == user_id)
        .order_by(RepairItem.status, RepairItem.priority.desc(), RepairItem.created_at.desc())
    ).all()
    revision_items = db.scalars(
        select(RevisionItem)
        .where(RevisionItem.user_id == user_id)
        .order_by(RevisionItem.status, RevisionItem.due_at, RevisionItem.created_at.desc())
    ).all()
    lesson_progress = db.scalars(
        select(LessonProgress)
        .where(LessonProgress.user_id == user_id)
        .order_by(LessonProgress.last_seen_at.desc())
    ).all()
    attempts = db.scalars(
        select(QuizAttempt)
        .where(QuizAttempt.user_id == user_id)
        .order_by(QuizAttempt.completed_at.desc())
    ).all()
    resources = db.scalars(
        select(GeneratedResource)
        .where(GeneratedResource.user_id == user_id)
        .order_by(GeneratedResource.created_at.desc())
    ).all()
    class_sessions = db.scalars(
        select(ClassSessionProgress)
        .where(ClassSessionProgress.user_id == user_id)
        .order_by(ClassSessionProgress.updated_at.desc())
        .limit(50)
    ).all()
    roadmaps = db.scalars(
        select(LearnAnythingRoadmap)
        .where(LearnAnythingRoadmap.user_id == user_id)
        .order_by(LearnAnythingRoadmap.updated_at.desc())
        .limit(50)
    ).all()
    chat_sessions = db.scalars(
        select(ChatSession)
        .where(ChatSession.user_id == user_id)
        .order_by(ChatSession.updated_at.desc())
        .limit(50)
    ).all()
    questions_asked = (
        db.scalar(
            select(func.count(ChatMessageRecord.id))
            .join(ChatSession, ChatMessageRecord.session_id == ChatSession.id)
            .where(
                ChatSession.user_id == user_id,
                ChatMessageRecord.role == "user",
            )
        )
        or 0
    )
    subject_map = {item.id: item.name for item in subjects}
    chapter_map = {item.id: item.title for item in chapters}
    chapter_by_id = {item.id: item for item in chapters}
    resolved_now = datetime.now(timezone.utc)
    today_plan_row = db.scalar(
        select(DailyPlan).where(
            DailyPlan.user_id == user_id,
            DailyPlan.plan_date == resolved_now.date(),
        )
    )
    today_plan_tasks = (
        list(
            db.scalars(
                select(DailyTask)
                .where(
                    DailyTask.user_id == user_id,
                    DailyTask.daily_plan_id == today_plan_row.id,
                )
                .order_by(DailyTask.priority.desc(), DailyTask.scheduled_for)
            )
        )
        if today_plan_row is not None
        else []
    )
    academic_state, next_action = build_academic_projection(
        profile=profile,
        subjects=list(subjects),
        chapters=list(chapters),
        tasks=today_plan_tasks,
        lesson_progress=list(lesson_progress),
        mastery=list(mastery),
        repairs=list(repair_items),
        attempts=list(attempts),
        class_sessions=list(class_sessions),
        roadmaps=list(roadmaps),
        chat_sessions=list(chat_sessions),
        now=resolved_now,
    )
    open_repair_keys = {
        item.concept_key
        for item in repair_items
        if item.status in {"open", "escalated"}
    }
    mastery_states = {
        item.topic_key: current_mastery_state(
            item,
            now=resolved_now,
            exam_date=profile.exam_date if profile else None,
            has_open_repair=item.topic_key in open_repair_keys,
        )
        for item in mastery
    }
    return LearningStateSummary(
        profile=LearningProfileOut.model_validate(profile, from_attributes=True) if profile else None,
        subjects=[LearningSubjectOut.model_validate(item, from_attributes=True) for item in subjects],
        chapters=[LearningChapterOut.model_validate(item, from_attributes=True) for item in chapters],
        tasks=[_task_out(item, subject_map, chapter_map) for item in tasks],
        today_plan=daily_plan_out(db, today_plan_row) if today_plan_row else None,
        lesson_progress=[
            LearningLessonProgressOut(
                id=item.id,
                chapter_id=item.chapter_id,
                chapter_catalog_id=chapter.catalog_id,
                chapter=chapter.title,
                subject_id=chapter.subject_id,
                mission_id=item.mission_id,
                status=item.status,
                progress_percent=item.progress_percent,
                current_step=item.current_step,
                completed_at=item.completed_at,
                last_seen_at=item.last_seen_at,
            )
            for item in lesson_progress
            if (chapter := chapter_by_id.get(item.chapter_id)) is not None
        ],
        mastery=[
            _mastery_out(item, resolved_state=mastery_states.get(item.topic_key))
            for item in mastery
        ],
        repair_items=[repair_item_out(item) for item in repair_items],
        revision_items=[_revision_item_out(item, subject_map, chapter_map) for item in revision_items],
        quiz_attempts=[
            LearningQuizAttemptOut(
                id=item.id,
                quiz_id=item.quiz_id,
                daily_task_id=item.daily_task_id,
                subject_id=item.subject_id,
                chapter_id=item.chapter_id,
                title=f"{chapter_map.get(item.chapter_id or '') or subject_map.get(item.subject_id or '') or 'Study'} assessment",
                subject=subject_map.get(item.subject_id or ""),
                chapter=chapter_map.get(item.chapter_id or ""),
                score=item.score,
                max_score=item.max_score,
                answers=item.answers,
                corrections=item.corrections,
                missing_keywords=item.missing_keywords,
                misconceptions=item.misconceptions,
                completed_at=item.completed_at,
            )
            for item in attempts
        ],
        resources=[
            LearningResourceOut(
                id=item.id,
                subject_id=item.subject_id,
                chapter_id=item.chapter_id,
                subject=subject_map.get(item.subject_id or ""),
                chapter=chapter_map.get(item.chapter_id or ""),
                source_id=item.source_id,
                resource_type=item.resource_type,
                title=item.title,
                status=item.status,
                storage_url=item.storage_url,
                resource_data=item.resource_data,
                created_at=item.created_at,
            )
            for item in resources
        ],
        completed_lessons=sum(item.status == "completed" for item in lesson_progress),
        completed_quiz_attempts=len(attempts),
        generated_resources=len(resources),
        generated_notes=sum(item.resource_type == "notes" for item in resources),
        questions_asked=int(questions_asked),
        academic_state=academic_state,
        next_action=next_action,
    )


def save_revision_item(
    db: Session,
    *,
    user_id: str,
    payload: RevisionItemSaveRequest,
) -> RevisionItemSaveResponse:
    """Create one student-requested revision item and its audit event.

    The client key is stable for the logical save. Replays therefore return the
    original item rather than adding a second reminder after a retry or route
    revisit.
    """
    existing = db.scalar(
        select(RevisionItem).where(
            RevisionItem.user_id == user_id,
            RevisionItem.client_item_id == payload.client_item_id,
        )
    )
    subjects, chapters = _task_maps(db, user_id)
    if existing is not None:
        return RevisionItemSaveResponse(
            item=_revision_item_out(existing, subjects, chapters),
            replayed=True,
            message="This revision item is already saved in your revision queue.",
        )

    chapter = db.scalar(
        select(Chapter).where(
            Chapter.user_id == user_id,
            Chapter.catalog_id == payload.chapter_catalog_id,
        )
    )
    if chapter is None:
        raise HTTPException(
            status_code=HTTP_422_UNPROCESSABLE_CONTENT,
            detail={
                "code": "CHAPTER_NOT_IN_STUDY_PLAN",
                "message": "This revision item belongs to a chapter that is not in the student's saved study plan.",
            },
        )

    now = datetime.now(timezone.utc)
    item = RevisionItem(
        user_id=user_id,
        client_item_id=payload.client_item_id,
        subject_id=chapter.subject_id,
        chapter_id=chapter.id,
        mission_id=payload.mission_id,
        topic_key=payload.topic_key,
        title=payload.title,
        source_kind=payload.source_kind,
        source_ref=payload.source_ref,
        status="pending",
        due_at=now + timedelta(days=payload.due_in_days),
        item_data={"chapter_catalog_id": payload.chapter_catalog_id},
    )
    db.add(item)
    db.flush()
    add_learning_event(
        db,
        user_id=user_id,
        event_type="REVISION_ITEM_CREATED",
        entity_type="revision_item",
        entity_id=item.id,
        idempotency_key=f"revision-item:{payload.client_item_id}",
        subject_id=item.subject_id,
        chapter_id=item.chapter_id,
        topic_key=item.topic_key,
        event_data={"source_kind": item.source_kind, "mission_id": item.mission_id},
    )
    try:
        db.commit()
    except IntegrityError:
        db.rollback()
        winner = db.scalar(
            select(RevisionItem).where(
                RevisionItem.user_id == user_id,
                RevisionItem.client_item_id == payload.client_item_id,
            )
        )
        if winner is None:
            raise
        subjects, chapters = _task_maps(db, user_id)
        return RevisionItemSaveResponse(
            item=_revision_item_out(winner, subjects, chapters),
            replayed=True,
            message="This revision item is already saved in your revision queue.",
        )
    db.refresh(item)
    if payload.due_in_days == 0:
        _replan_if_setup_exists(
            db,
            user_id=user_id,
            idempotency_key=f"revision-created:{item.id}",
            reason="revision_due",
        )
    subjects, chapters = _task_maps(db, user_id)
    return RevisionItemSaveResponse(
        item=_revision_item_out(item, subjects, chapters),
        message="Saved for revision. It now appears in your account revision queue.",
    )


def record_revision_result(
    db: Session,
    *,
    user_id: str,
    revision_item_id: str,
    payload: RevisionResultRequest,
) -> RevisionResultResponse:
    item = db.scalar(
        select(RevisionItem).where(
            RevisionItem.id == revision_item_id,
            RevisionItem.user_id == user_id,
        ).with_for_update()
    )
    if item is None:
        raise HTTPException(status_code=404, detail="Revision item not found.")
    event_key = f"revision-result:{payload.client_event_id}"
    from app.models.learning_state import LearningEvent

    replay = db.scalar(
        select(LearningEvent).where(
            LearningEvent.user_id == user_id,
            LearningEvent.idempotency_key == event_key,
        )
    )
    subjects, chapters = _task_maps(db, user_id)
    if replay is not None:
        return RevisionResultResponse(
            item=_revision_item_out(item, subjects, chapters),
            replayed=True,
            message="This revision result was already saved.",
        )

    now = datetime.now(timezone.utc)
    metadata = dict(item.item_data or {})
    successes = int(metadata.get("successful_reviews", 0) or 0)
    if payload.correct:
        successes += 1
        if successes >= 4:
            item.status = "completed"
            item.completed_at = now
            item.due_at = None
        else:
            interval_days = (3, 7, 14)[successes - 1]
            item.status = "pending"
            item.due_at = now + timedelta(days=interval_days)
    else:
        successes = max(0, successes - 1)
        item.status = "pending"
        item.completed_at = None
        item.due_at = now + timedelta(days=1)
    item.item_data = {
        **metadata,
        "successful_reviews": successes,
        "last_result": "correct" if payload.correct else "incorrect",
        "last_reviewed_at": now.isoformat(),
    }

    today_plan = db.scalar(
        select(DailyPlan).where(
            DailyPlan.user_id == user_id,
            DailyPlan.plan_date == now.date(),
        )
    )
    if today_plan is not None:
        for task in db.scalars(
            select(DailyTask).where(
                DailyTask.user_id == user_id,
                DailyTask.daily_plan_id == today_plan.id,
                DailyTask.status == "pending",
            )
        ):
            if (task.task_metadata or {}).get("revision_item_id") != item.id:
                continue
            task.status = "completed"
            task.completed_at = now
            task.task_metadata = {
                **dict(task.task_metadata or {}),
                "completion_evidence": {
                    "kind": "revision_result",
                    "client_event_id": payload.client_event_id,
                    "correct": payload.correct,
                },
            }
            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:revision:{payload.client_event_id}:{task.id}",
                subject_id=task.subject_id,
                chapter_id=task.chapter_id,
                topic_key=item.topic_key,
                event_data={"daily_plan_id": task.daily_plan_id, "evidence_type": "revision_result"},
            )
        refresh_daily_plan_totals(db, plan_id=today_plan.id, user_id=user_id)

    add_learning_event(
        db,
        user_id=user_id,
        event_type="REVISION_COMPLETED" if payload.correct else "REVISION_RESCHEDULED",
        entity_type="revision_item",
        entity_id=item.id,
        idempotency_key=event_key,
        subject_id=item.subject_id,
        chapter_id=item.chapter_id,
        topic_key=item.topic_key,
        event_data={
            "correct": payload.correct,
            "successful_reviews": successes,
            "next_due_at": item.due_at.isoformat() if item.due_at else None,
            "status": item.status,
        },
    )
    db.commit()
    _replan_if_setup_exists(
        db,
        user_id=user_id,
        idempotency_key=f"revision:{payload.client_event_id}",
        reason="meaningful_evidence",
        now=now,
    )
    subjects, chapters = _task_maps(db, user_id)
    return RevisionResultResponse(
        item=_revision_item_out(item, subjects, chapters),
        message=(
            "Revision secured. This item is complete."
            if item.status == "completed"
            else "Revision saved. DocDoe scheduled the next transparent recall window."
        ),
    )


def _study_followup_out(item: StudyFollowUp) -> StudyFollowUpOut:
    return StudyFollowUpOut(
        id=item.id,
        client_followup_id=item.client_followup_id,
        chat_session_id=item.chat_session_id,
        kind=item.kind,
        title=item.title,
        subject=item.subject,
        chapter=item.chapter,
        topic=item.topic,
        status=item.status,
        due_at=item.due_at,
        estimated_minutes=item.estimated_minutes,
        source_ref=item.source_ref,
        target=dict(item.target_data or {}),
        created_at=item.created_at,
    )


def create_study_followup(
    db: Session,
    *,
    user_id: str,
    payload: StudyFollowUpCreateRequest,
) -> StudyFollowUpOut:
    existing = db.scalar(
        select(StudyFollowUp).where(
            StudyFollowUp.user_id == user_id,
            StudyFollowUp.client_followup_id == payload.client_followup_id,
        )
    )
    if existing is not None:
        return _study_followup_out(existing)
    session = db.scalar(
        select(ChatSession).where(
            ChatSession.id == payload.chat_session_id,
            ChatSession.user_id == user_id,
        )
    )
    if session is None:
        raise HTTPException(status_code=404, detail="Study Chat session not found.")
    if payload.source_message_id:
        message = db.scalar(
            select(ChatMessageRecord).where(
                ChatMessageRecord.id == payload.source_message_id,
                ChatMessageRecord.session_id == session.id,
            )
        )
        if message is None:
            raise HTTPException(status_code=404, detail="Study Chat message not found.")
    safe_href = f"/study-chat?sessionId={session.id}"
    item = StudyFollowUp(
        user_id=user_id,
        chat_session_id=session.id,
        source_message_id=payload.source_message_id,
        client_followup_id=payload.client_followup_id,
        kind=payload.kind,
        title=payload.title,
        subject=payload.subject,
        chapter=payload.chapter,
        topic=payload.topic,
        status="open",
        due_at=payload.due_at or datetime.now(timezone.utc),
        estimated_minutes=payload.estimated_minutes,
        source_ref=payload.source_ref,
        target_data={**payload.target, "href": safe_href, "session_id": session.id},
    )
    db.add(item)
    db.flush()
    add_learning_event(
        db,
        user_id=user_id,
        event_type="CHAT_FOLLOWUP_CREATED",
        entity_type="study_followup",
        entity_id=item.id,
        idempotency_key=f"chat-followup:{payload.client_followup_id}",
        event_data={"chat_session_id": session.id, "kind": item.kind},
    )
    db.commit()
    _replan_if_setup_exists(
        db,
        user_id=user_id,
        idempotency_key=f"chat-followup:{item.id}",
        reason="meaningful_evidence",
    )
    db.refresh(item)
    return _study_followup_out(item)


def set_study_followup_status(
    db: Session,
    *,
    user_id: str,
    followup_id: str,
    new_status: str,
) -> StudyFollowUpOut:
    item = db.scalar(
        select(StudyFollowUp).where(
            StudyFollowUp.id == followup_id,
            StudyFollowUp.user_id == user_id,
        ).with_for_update()
    )
    if item is None:
        raise HTTPException(status_code=404, detail="Study follow-up not found.")
    if item.status == new_status:
        return _study_followup_out(item)
    item.status = new_status
    item.completed_at = datetime.now(timezone.utc) if new_status == "resolved" else None
    add_learning_event(
        db,
        user_id=user_id,
        event_type="CHAT_FOLLOWUP_RESOLVED" if new_status == "resolved" else "CHAT_FOLLOWUP_DISMISSED",
        entity_type="study_followup",
        entity_id=item.id,
        idempotency_key=f"chat-followup-status:{item.id}:{new_status}",
        event_data={"chat_session_id": item.chat_session_id},
    )
    db.commit()
    _replan_if_setup_exists(
        db,
        user_id=user_id,
        idempotency_key=f"chat-followup-status:{item.id}:{new_status}",
        reason="meaningful_evidence",
    )
    return _study_followup_out(item)


def list_open_study_followups(db: Session, *, user_id: str) -> list[StudyFollowUpOut]:
    items = db.scalars(
        select(StudyFollowUp)
        .where(StudyFollowUp.user_id == user_id, StudyFollowUp.status == "open")
        .order_by(StudyFollowUp.due_at, StudyFollowUp.created_at)
    )
    return [_study_followup_out(item) for item in items]


def sync_class_session_lesson_progress(
    db: Session,
    *,
    user_id: str,
    class_session_id: str,
    data: dict,
) -> None:
    """Promote an in-progress Tuition class into LessonProgress so the planner can continue it."""

    parts = class_session_id.split("|", 3)
    if len(parts) < 4:
        return
    catalog_id, mission_id = parts[2], parts[3]
    if not catalog_id or not mission_id or not data.get("started"):
        return
    chapter = db.scalar(
        select(Chapter).where(Chapter.user_id == user_id, Chapter.catalog_id == catalog_id)
    )
    if chapter is None:
        return
    progress = db.scalar(
        select(LessonProgress).where(
            LessonProgress.user_id == user_id,
            LessonProgress.chapter_id == chapter.id,
            LessonProgress.mission_id == mission_id,
        )
    )
    if progress is not None and progress.status == "completed":
        return
    step_id = str(data.get("activeStepId") or "intro")
    unlocked = data.get("maxUnlockedIndex")
    step_index = int(unlocked) if isinstance(unlocked, (int, float)) else 0
    percent = min(99, max(8, (step_index + 1) * 8))
    now = datetime.now(timezone.utc)
    if progress is None:
        progress = LessonProgress(
            user_id=user_id,
            chapter_id=chapter.id,
            mission_id=mission_id,
            status="in_progress",
            progress_percent=percent,
            current_step=max(1, step_index + 1),
            last_seen_at=now,
        )
        db.add(progress)
    else:
        progress.status = "in_progress"
        progress.progress_percent = percent
        progress.current_step = max(1, step_index + 1)
        progress.last_seen_at = now
    db.flush()
    _replan_if_setup_exists(
        db,
        user_id=user_id,
        idempotency_key=f"class-session:{class_session_id}:{step_id}:{step_index}",
        reason="meaningful_evidence",
        now=now,
    )


def delete_generated_resource(db: Session, *, user_id: str, resource_id: str) -> None:
    resource = db.scalar(
        select(GeneratedResource).where(
            GeneratedResource.id == resource_id,
            GeneratedResource.user_id == user_id,
        )
    )
    if resource is None:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail={
                "code": "LEARNING_RESOURCE_NOT_FOUND",
                "message": "This saved resource was not found.",
            },
        )
    db.delete(resource)
    db.commit()


def _rebalance_tasks(plan: StudyPlan, tasks: Iterable[DailyTask], *, start_day: date) -> list[DailyTask]:
    day = start_day
    used_minutes = 0
    changed: list[DailyTask] = []
    for task in tasks:
        if used_minutes and used_minutes + task.duration_minutes > plan.daily_minutes:
            day += timedelta(days=1)
            used_minutes = 0
        next_time = _scheduled_datetime(day, "evening") + timedelta(minutes=used_minutes)
        if task.scheduled_for != next_time:
            task.scheduled_for = next_time
            changed.append(task)
        used_minutes += task.duration_minutes
    return changed


def update_task_status(
    db: Session,
    *,
    user_id: str,
    task_id: str,
    new_status: str,
    reason: str | None,
) -> TaskStatusResponse:
    task = db.scalar(
        select(DailyTask).where(DailyTask.id == task_id, DailyTask.user_id == user_id).with_for_update()
    )
    if task is None:
        raise HTTPException(status_code=404, detail="Study task not found.")
    plan = db.scalar(select(StudyPlan).where(StudyPlan.id == task.study_plan_id, StudyPlan.user_id == user_id))
    if plan is None:
        raise HTTPException(status_code=404, detail="Study plan not found.")

    recalculated: list[DailyTask] = []
    now = datetime.now(timezone.utc)
    if new_status == "pending":
        if task.task_type not in {"chat_follow_up", "recap", "upload", "note"}:
            raise HTTPException(
                status_code=status.HTTP_409_CONFLICT,
                detail={
                    "code": "EVIDENCE_REQUIRED",
                    "message": "This learning task changes only when its lesson or practice evidence changes.",
                },
            )
        task.status = "pending"
        task.completed_at = None
        message = "Task moved back to pending."
    elif new_status == "completed":
        if task.task_type not in {"chat_follow_up", "recap", "upload", "note"}:
            raise HTTPException(
                status_code=status.HTTP_409_CONFLICT,
                detail={
                    "code": "EVIDENCE_REQUIRED",
                    "message": "Finish the linked lesson, quiz, revision, or repair so DocDoe can verify this task.",
                },
            )
        task.status = "completed"
        task.completed_at = now
        message = "Follow-up completed. Your saved plan now reflects it."
        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:explicit:{task.id}",
            subject_id=task.subject_id,
            chapter_id=task.chapter_id,
            event_data={"daily_plan_id": task.daily_plan_id, "evidence_type": "explicit_followup"},
        )
    elif new_status == "skipped":
        task.status = "skipped"
        task.task_metadata = {**(task.task_metadata or {}), "skip_reason": reason}
        message = "Task skipped. The remaining plan was kept unchanged."
    else:
        task.status = "missed"
        task.task_metadata = {**(task.task_metadata or {}), "miss_reason": reason}
        replacement = DailyTask(
            user_id=user_id,
            study_plan_id=task.study_plan_id,
            subject_id=task.subject_id,
            chapter_id=task.chapter_id,
            task_type=task.task_type,
            title=task.title,
            status="pending",
            scheduled_for=now + timedelta(days=1),
            duration_minutes=task.duration_minutes,
            priority=min(2.0, task.priority + 0.25),
            href=task.href,
            mission_id=task.mission_id,
            rescheduled_from_id=task.id,
            task_metadata={**(task.task_metadata or {}), "recalculated": True},
        )
        db.add(replacement)
        db.flush()
        pending = db.scalars(
            select(DailyTask).where(
                DailyTask.user_id == user_id,
                DailyTask.study_plan_id == plan.id,
                DailyTask.status == "pending",
            ).order_by(DailyTask.priority.desc(), DailyTask.scheduled_for)
        ).all()
        recalculated = _rebalance_tasks(plan, pending, start_day=date.today() + timedelta(days=1))
        if replacement not in recalculated:
            recalculated.insert(0, replacement)
        message = "Missed task rescheduled. DocDoe recalculated the upcoming plan instead of leaving it late."

    db.add(UsageEvent(user_id=user_id, event_type=f"task_{new_status}", resource_type="daily_task", event_data={"task_id": task.id}))
    refresh_daily_plan_totals(db, plan_id=task.daily_plan_id, user_id=user_id)
    db.commit()
    if task.daily_plan_id and new_status in {"completed", "missed", "skipped"}:
        _replan_if_setup_exists(
            db,
            user_id=user_id,
            idempotency_key=f"task-status:{task.id}:{new_status}:{date.today().isoformat()}",
            reason="meaningful_evidence",
        )
    subject_map, chapter_map = _task_maps(db, user_id)
    return TaskStatusResponse(
        task=_task_out(task, subject_map, chapter_map),
        recalculated_tasks=[_task_out(item, subject_map, chapter_map) for item in recalculated],
        message=message,
    )


def record_lesson_progress(
    db: Session,
    *,
    user_id: str,
    payload: LessonProgressRequest,
) -> LessonProgressResponse:
    chapter = db.scalar(
        select(Chapter).where(
            Chapter.user_id == user_id,
            Chapter.catalog_id == payload.chapter_catalog_id,
        )
    )
    if chapter is None:
        raise HTTPException(
            status_code=HTTP_422_UNPROCESSABLE_CONTENT,
            detail={
                "code": "CHAPTER_NOT_IN_STUDY_PLAN",
                "message": "This chapter is not in the student's saved study plan.",
            },
        )

    task: DailyTask | None = None
    if payload.daily_task_id:
        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=404, detail="Study task not found.")
        if task.chapter_id and task.chapter_id != chapter.id:
            raise HTTPException(
                status_code=HTTP_422_UNPROCESSABLE_CONTENT,
                detail={
                    "code": "TASK_CHAPTER_MISMATCH",
                    "message": "This task belongs to a different chapter.",
                },
            )
        if task.mission_id and task.mission_id != payload.mission_id:
            raise HTTPException(
                status_code=HTTP_422_UNPROCESSABLE_CONTENT,
                detail={
                    "code": "TASK_MISSION_MISMATCH",
                    "message": "This task belongs to a different lesson mission.",
                },
            )
    else:
        task = db.scalar(
            select(DailyTask).where(
                DailyTask.user_id == user_id,
                DailyTask.chapter_id == chapter.id,
                DailyTask.mission_id == payload.mission_id,
                DailyTask.status == "pending",
            ).order_by(DailyTask.scheduled_for)
        )

    progress = db.scalar(
        select(LessonProgress).where(
            LessonProgress.user_id == user_id,
            LessonProgress.chapter_id == chapter.id,
            LessonProgress.mission_id == payload.mission_id,
        ).with_for_update()
    )
    was_completed = progress is not None and progress.status == "completed"
    now = datetime.now(timezone.utc)
    if progress is None:
        progress = LessonProgress(
            user_id=user_id,
            chapter_id=chapter.id,
            mission_id=payload.mission_id,
        )
        db.add(progress)

    progress.status = payload.status
    progress.progress_percent = max(progress.progress_percent or 0, payload.progress_percent)
    progress.current_step = max(progress.current_step or 0, payload.current_step)
    progress.last_seen_at = now
    if payload.status == "completed":
        progress.progress_percent = 100
        progress.completed_at = progress.completed_at or now
        if task is not None and task.task_type in {"lesson", "new_lesson", "continue_lesson"}:
            task.status = "completed"
            task.completed_at = task.completed_at or now

    profile = db.scalar(
        select(StudentProfileState).where(StudentProfileState.user_id == user_id)
    )
    if profile is not None:
        profile.current_subject_id = chapter.subject_id
        profile.current_chapter_id = chapter.id
        profile.current_mission_id = payload.mission_id

    if payload.status == "completed" and not was_completed:
        duration = max(0, payload.duration_minutes)
        db.add(
            StudySession(
                user_id=user_id,
                daily_task_id=task.id if task else None,
                subject_id=chapter.subject_id,
                chapter_id=chapter.id,
                status="completed",
                started_at=now - timedelta(minutes=duration),
                ended_at=now,
                duration_minutes=duration,
                session_data={
                    "mission_id": payload.mission_id,
                    "title": payload.title,
                    "source": "tuition_class",
                },
            )
        )
        db.add(
            UsageEvent(
                user_id=user_id,
                event_type="lesson_completed",
                resource_type="lesson_progress",
                event_data={
                    "chapter_id": chapter.id,
                    "mission_id": payload.mission_id,
                    "daily_task_id": task.id if task else None,
                },
            )
        )
        # The lesson row, linked task, study session and event commit together.
        # A retry sees the completed lesson and does not create a second session
        # or completion event.
        db.flush()
        add_learning_event(
            db,
            user_id=user_id,
            event_type="LESSON_COMPLETED",
            entity_type="lesson_progress",
            entity_id=progress.id,
            idempotency_key=f"lesson-completed:{progress.id}",
            subject_id=chapter.subject_id,
            chapter_id=chapter.id,
            topic_key=f"{chapter.catalog_id}:{payload.mission_id}",
            event_data={"mission_id": payload.mission_id, "daily_task_id": task.id if task else None},
        )
        if task is not None and task.status == "completed":
            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:lesson:{progress.id}:{task.id}",
                subject_id=task.subject_id,
                chapter_id=task.chapter_id,
                topic_key=f"{chapter.catalog_id}:{payload.mission_id}",
                event_data={"daily_plan_id": task.daily_plan_id, "evidence_type": "lesson_progress"},
            )
            refresh_daily_plan_totals(db, plan_id=task.daily_plan_id, user_id=user_id)

    db.commit()
    if payload.status == "in_progress" or (payload.status == "completed" and not was_completed):
        _replan_if_setup_exists(
            db,
            user_id=user_id,
            idempotency_key=f"lesson:{progress.id}:{payload.status}:{payload.progress_percent}",
            reason="meaningful_evidence",
            now=now,
        )
    db.refresh(progress)
    completed_lessons = len(
        db.scalars(
            select(LessonProgress).where(
                LessonProgress.user_id == user_id,
                LessonProgress.status == "completed",
            )
        ).all()
    )
    return LessonProgressResponse(
        progress_id=progress.id,
        status=progress.status,
        progress_percent=progress.progress_percent,
        completed_lessons=completed_lessons,
        completed_task_id=task.id if task and task.status == "completed" else None,
        message=(
            "Class completed. The lesson, study session, and today's plan now use this result."
            if payload.status == "completed"
            else "Class progress saved."
        ),
    )


def _replayed_assessment_response(
    db: Session,
    *,
    user_id: str,
    payload: AssessmentResultRequest,
    attempt: QuizAttempt,
) -> AssessmentConsequenceResponse:
    mastery = db.scalar(
        select(TopicMastery).where(
            TopicMastery.user_id == user_id,
            TopicMastery.topic_key == payload.topic_key,
        )
    )
    score = mastery.score if mastery is not None else 0.0
    tasks = db.scalars(
        select(DailyTask).where(DailyTask.user_id == user_id)
    ).all()
    revision_task = next(
        (
            task
            for task in tasks
            if (task.task_metadata or {}).get("assessment_attempt_id") == attempt.id
        ),
        None,
    )
    next_task = revision_task or next(
        (
            task
            for task in sorted(
                tasks,
                key=lambda item: (-item.priority, item.scheduled_for),
            )
            if task.status == "pending"
        ),
        None,
    )
    subject_map, chapter_map = _task_maps(db, user_id)
    return AssessmentConsequenceResponse(
        attempt_id=attempt.id,
        mastery_before=round(score, 2),
        mastery_after=round(score, 2),
        revision_task=(
            _task_out(revision_task, subject_map, chapter_map)
            if revision_task
            else None
        ),
        next_recommended_task_id=next_task.id if next_task else None,
        message="This assessment was already saved. No duplicate mastery or revision change was created.",
    )


def record_assessment(
    db: Session,
    *,
    user_id: str,
    payload: AssessmentResultRequest,
) -> AssessmentConsequenceResponse:
    from app.services.mastery_engine import EvidenceEvent, MasterySnapshot, apply_evidence

    if payload.client_attempt_id:
        existing_attempt = db.scalar(
            select(QuizAttempt).where(
                QuizAttempt.user_id == user_id,
                QuizAttempt.client_attempt_id == payload.client_attempt_id,
            )
        )
        if existing_attempt is not None:
            return _replayed_assessment_response(
                db,
                user_id=user_id,
                payload=payload,
                attempt=existing_attempt,
            )

    observed = max(0.0, min(100.0, payload.score / payload.max_score * 100.0))
    now = datetime.now(timezone.utc)
    profile = db.scalar(select(StudentProfileState).where(StudentProfileState.user_id == user_id))
    exam_date = profile.exam_date if profile is not None else None
    mastery = db.scalar(
        select(TopicMastery).where(
            TopicMastery.user_id == user_id,
            TopicMastery.topic_key == payload.topic_key,
        ).with_for_update()
    )
    before = mastery.score if mastery else 0.0

    if mastery is None:
        mastery = TopicMastery(
            user_id=user_id,
            subject_id=payload.subject_id,
            chapter_id=payload.chapter_id,
            topic_key=payload.topic_key,
            topic_label=payload.topic_label,
        )
        db.add(mastery)
    mastery.subject_id = payload.subject_id or mastery.subject_id
    mastery.chapter_id = payload.chapter_id or mastery.chapter_id
    mastery.topic_label = payload.topic_label

    if payload.mastery_already_recorded:
        # Per-question checkpoint evidence already updated mastery during the
        # class; this call only stores the attempt record and consequences.
        after = mastery.score
    else:
        # Route every answer through the central mastery engine so evidence
        # weighting, repeat-question dampening, states, and revision intervals
        # have exactly one implementation (and the evidence log is preserved).
        # A freshly created ORM row has None attributes until flush — coalesce.
        snapshot = MasterySnapshot(
            score=mastery.score or 0.0,
            confidence=mastery.confidence or 0.0,
            attempts_count=mastery.attempts_count or 0,
            state=mastery.last_result or "not_started",
            next_review_at=mastery.next_review_at,
            evidence=dict(mastery.evidence or {}),
        )
        answer_events = [
            EvidenceEvent(
                kind="quiz_aggregate",
                correct=bool(answer.get("correct")),
                at=now,
                question_id=str(answer.get("question_id")) if answer.get("question_id") else None,
                source="assessment",
            )
            for answer in payload.answers
        ] or [
            EvidenceEvent(kind="quiz_aggregate", correct=observed >= 60.0, at=now, source="assessment")
        ]
        for event in answer_events:
            apply_evidence(snapshot, event, exam_date=exam_date, has_open_repair=False)
        mastery.score = snapshot.score
        mastery.confidence = snapshot.confidence
        mastery.attempts_count = snapshot.attempts_count
        mastery.last_result = snapshot.state
        mastery.next_review_at = snapshot.next_review_at
        mastery.evidence = snapshot.evidence
        after = snapshot.score

    attempt = QuizAttempt(
        user_id=user_id,
        client_attempt_id=payload.client_attempt_id,
        quiz_id=payload.quiz_id,
        daily_task_id=payload.daily_task_id,
        subject_id=payload.subject_id,
        chapter_id=payload.chapter_id,
        score=payload.score,
        max_score=payload.max_score,
        answers=payload.answers,
        corrections=payload.corrections,
        missing_keywords=payload.missing_keywords,
        misconceptions=payload.misconceptions,
        completed_at=now,
    )
    db.add(attempt)
    try:
        db.flush()
    except IntegrityError:
        db.rollback()
        if payload.client_attempt_id:
            existing_attempt = db.scalar(
                select(QuizAttempt).where(
                    QuizAttempt.user_id == user_id,
                    QuizAttempt.client_attempt_id == payload.client_attempt_id,
                )
            )
            if existing_attempt is not None:
                return _replayed_assessment_response(
                    db,
                    user_id=user_id,
                    payload=payload,
                    attempt=existing_attempt,
                )
        raise

    assessed_task: DailyTask | None = None
    if payload.daily_task_id:
        assessed_task = db.scalar(
            select(DailyTask).where(DailyTask.id == payload.daily_task_id, DailyTask.user_id == user_id)
        )
        if assessed_task is None:
            raise HTTPException(status_code=404, detail="Study task not found.")
        if payload.chapter_id and assessed_task.chapter_id and assessed_task.chapter_id != payload.chapter_id:
            raise HTTPException(
                status_code=status.HTTP_409_CONFLICT,
                detail={"code": "TASK_EVIDENCE_MISMATCH", "message": "This quiz belongs to a different study task."},
            )
        if assessed_task.task_type in {
            "quiz",
            "pyq_practice",
            "board_answer_practice",
            "test",
            "practice",
        }:
            assessed_task.status = "completed"
            assessed_task.completed_at = now
            assessed_task.task_metadata = {
                **dict(assessed_task.task_metadata or {}),
                "completion_evidence": {"kind": "quiz_attempt", "attempt_id": attempt.id},
            }
            add_learning_event(
                db,
                user_id=user_id,
                event_type="TASK_COMPLETED",
                entity_type="daily_task",
                entity_id=assessed_task.id,
                idempotency_key=f"task-completed:quiz:{attempt.id}:{assessed_task.id}",
                subject_id=assessed_task.subject_id,
                chapter_id=assessed_task.chapter_id,
                topic_key=payload.topic_key,
                event_data={"daily_plan_id": assessed_task.daily_plan_id, "evidence_type": "quiz_attempt"},
            )
            refresh_daily_plan_totals(
                db,
                plan_id=assessed_task.daily_plan_id,
                user_id=user_id,
            )

    revision_item: RevisionItem | None = None
    # Aggregate quiz evidence creates one durable revision item. The Learning
    # Engine, not this mutation endpoint, decides when that item fits a day.
    if (
        not payload.mastery_already_recorded
        and (after < 75 or payload.missing_keywords or payload.misconceptions)
    ):
        review_days = 7 if after >= 85 else 3 if after >= 70 else 1
        mission_id = payload.mission_id
        if mission_id is None and ":" in payload.topic_key:
            inferred_mission_id = payload.topic_key.rsplit(":", 1)[-1]
            if inferred_mission_id.startswith("M") and inferred_mission_id[1:].isdigit():
                mission_id = inferred_mission_id
        due_immediately = after < 50
        revision_item = RevisionItem(
            user_id=user_id,
            client_item_id=f"assessment:{attempt.id}",
            subject_id=payload.subject_id,
            chapter_id=payload.chapter_id,
            mission_id=mission_id,
            topic_key=payload.topic_key,
            title=payload.topic_label,
            source_kind="lesson_recap",
            source_ref=attempt.id,
            status="due" if due_immediately else "pending",
            due_at=now if due_immediately else now + timedelta(days=review_days),
            item_data={
                "chapter_catalog_id": (
                    db.scalar(select(Chapter.catalog_id).where(Chapter.id == payload.chapter_id))
                    if payload.chapter_id
                    else None
                ),
                "assessment_attempt_id": attempt.id,
                "missing_keywords": payload.missing_keywords,
                "misconceptions": payload.misconceptions,
                "successful_reviews": 0,
                "estimated_minutes": 10,
            },
        )
        db.add(revision_item)
        db.flush()
        add_learning_event(
            db,
            user_id=user_id,
            event_type="REVISION_ITEM_CREATED",
            entity_type="revision_item",
            entity_id=revision_item.id,
            idempotency_key=f"revision-item:assessment:{attempt.id}",
            subject_id=revision_item.subject_id,
            chapter_id=revision_item.chapter_id,
            topic_key=revision_item.topic_key,
            event_data={"source_kind": "assessment", "due_at": revision_item.due_at.isoformat()},
        )

    db.add(UsageEvent(user_id=user_id, event_type="assessment_completed", resource_type="quiz_attempt", event_data={"attempt_id": attempt.id, "mastery_after": after}))
    add_learning_event(
        db,
        user_id=user_id,
        event_type="QUIZ_COMPLETED",
        entity_type="quiz_attempt",
        entity_id=attempt.id,
        idempotency_key=f"quiz-completed:{attempt.id}",
        subject_id=attempt.subject_id,
        chapter_id=attempt.chapter_id,
        topic_key=payload.topic_key,
        event_data={"quiz_id": attempt.quiz_id, "score": attempt.score, "max_score": attempt.max_score},
    )
    db.commit()

    _replan_if_setup_exists(
        db,
        user_id=user_id,
        idempotency_key=f"assessment:{attempt.id}",
        reason="meaningful_evidence",
        now=now,
    )
    current_daily_plan = db.scalar(
        select(DailyPlan).where(
            DailyPlan.user_id == user_id,
            DailyPlan.plan_date == now.date(),
        )
    )
    next_task = (
        db.scalar(
            select(DailyTask)
            .where(
                DailyTask.user_id == user_id,
                DailyTask.daily_plan_id == current_daily_plan.id,
                DailyTask.status == "pending",
            )
            .order_by(DailyTask.priority.desc(), DailyTask.scheduled_for)
        )
        if current_daily_plan
        else None
    )
    planned_revision = None
    if current_daily_plan is not None and revision_item is not None:
        planned_revision = next(
            (
                task
                for task in db.scalars(
                    select(DailyTask).where(
                        DailyTask.user_id == user_id,
                        DailyTask.daily_plan_id == current_daily_plan.id,
                        DailyTask.status == "pending",
                    )
                )
                if (task.task_metadata or {}).get("revision_item_id") == revision_item.id
                or (task.task_metadata or {}).get("assessment_attempt_id") == attempt.id
            ),
            None,
        )

    subject_map, chapter_map = _task_maps(db, user_id)
    return AssessmentConsequenceResponse(
        attempt_id=attempt.id,
        mastery_before=round(before, 2),
        mastery_after=round(after, 2),
        revision_task=(
            _task_out(planned_revision, subject_map, chapter_map)
            if planned_revision
            else None
        ),
        next_recommended_task_id=(
            planned_revision.id if planned_revision is not None else (next_task.id if next_task else None)
        ),
        message=(
            "A revision item was saved because this answer exposed a weak concept."
            if revision_item
            else "Mastery improved and the next planned lesson remains recommended."
        ),
    )


def adjust_plan_from_assistant(
    db: Session,
    *,
    user_id: str,
    payload: PlanAdjustmentRequest,
) -> PlanAdjustmentResponse:
    """Persist an explicit focus, then let the central engine replan."""

    profile = db.scalar(
        select(StudentProfileState)
        .where(StudentProfileState.user_id == user_id)
        .with_for_update()
    )
    active_plan = db.scalar(
        select(StudyPlan)
        .where(StudyPlan.user_id == user_id, StudyPlan.status == "active")
        .order_by(StudyPlan.created_at.desc())
    )
    if profile is None or active_plan is None:
        raise HTTPException(
            status_code=status.HTTP_409_CONFLICT,
            detail={
                "code": "STUDY_PLAN_REQUIRED",
                "message": "Complete your study setup before DocDoe changes the plan.",
            },
        )

    subjects = list(
        db.scalars(
            select(Subject).where(
                Subject.user_id == user_id,
                Subject.status == "active",
            )
        )
    )
    target = next(
        (item for item in subjects if item.name.casefold() == payload.target_subject.casefold()),
        None,
    )
    if target is None:
        raise HTTPException(
            status_code=HTTP_422_UNPROCESSABLE_CONTENT,
            detail={
                "code": "SUBJECT_NOT_SELECTED",
                "message": f"{payload.target_subject} is not in your selected subjects.",
            },
        )

    from app.services.content_manifest import load_content_manifest

    manifest = load_content_manifest()
    available = (
        list(manifest.available_for_subject(target.name))
        if manifest.supports_curriculum(profile.board, profile.class_level)
        else []
    )
    by_number = {str(index + 1): chapter for index, chapter in enumerate(available)}
    by_title = {chapter.title.casefold(): chapter for chapter in available}
    resolved = []
    unavailable_chapters: list[str] = []
    for label in payload.chapters:
        normalized = label.casefold().removeprefix("chapter ").strip()
        chapter = by_number.get(normalized) or by_title.get(normalized)
        if chapter is None:
            unavailable_chapters.append(label)
        elif all(existing.chapter_id != chapter.chapter_id for existing in resolved):
            resolved.append(chapter)

    if not resolved:
        raise HTTPException(
            status_code=HTTP_422_UNPROCESSABLE_CONTENT,
            detail={
                "code": "CHAPTER_NOT_AVAILABLE",
                "message": (
                    f"DocDoe has no verified {target.name} lesson for "
                    f"{', '.join(unavailable_chapters)}. Your plan is unchanged."
                ),
                "unavailable_chapters": unavailable_chapters,
            },
        )

    replacement = next(
        (
            item
            for item in subjects
            if payload.replace_subject
            and item.name.casefold() == payload.replace_subject.casefold()
            and item.id != target.id
        ),
        None,
    )
    ordered = [target]
    ordered.extend(
        item
        for item in sorted(subjects, key=lambda value: (value.priority, value.name))
        if item.id != target.id and (replacement is None or item.id != replacement.id)
    )
    if replacement is not None:
        ordered.append(replacement)
    for priority, subject in enumerate(ordered):
        subject.priority = priority

    today = datetime.now(timezone.utc).date()
    focus_key = f"{today.isoformat()}:{target.id}:{','.join(item.chapter_id for item in resolved)}"
    profile.preferences = {
        **dict(profile.preferences or {}),
        "today_focus": {
            "date": today.isoformat(),
            "subject_id": target.id,
            "subject": target.name,
            "chapter_catalog_ids": [item.chapter_id for item in resolved],
            "reason": payload.reason,
        },
    }
    profile.current_subject_id = target.id
    first_chapter = db.scalar(
        select(Chapter).where(
            Chapter.user_id == user_id,
            Chapter.subject_id == target.id,
            Chapter.catalog_id == resolved[0].chapter_id,
        )
    )
    if first_chapter is not None:
        profile.current_chapter_id = first_chapter.id

    add_learning_event(
        db,
        user_id=user_id,
        event_type="PLAN_PREFERENCE_CHANGED",
        entity_type="student_profile",
        entity_id=profile.id,
        idempotency_key=f"today-focus:{focus_key}",
        subject_id=target.id,
        chapter_id=first_chapter.id if first_chapter else None,
        event_data={
            "date": today.isoformat(),
            "chapter_catalog_ids": [item.chapter_id for item in resolved],
            "replaced_subject_id": replacement.id if replacement else None,
        },
    )
    db.add(
        UsageEvent(
            user_id=user_id,
            event_type="assistant_plan_adjusted",
            resource_type="daily_plan",
            event_data={
                "target_subject": target.name,
                "chapter_catalog_ids": [item.chapter_id for item in resolved],
            },
        )
    )
    db.commit()

    result = replan_today(
        db,
        user_id=user_id,
        idempotency_key=f"explicit-focus:{focus_key}",
        reason="explicit_student_focus",
    )
    message = f"Plan updated. {target.name} is now prioritised by the central Today Plan."
    if unavailable_chapters:
        message += (
            f" I left out {', '.join(unavailable_chapters)} because no verified lesson exists."
        )
    return PlanAdjustmentResponse(
        tasks=result.tasks,
        unavailable_chapters=unavailable_chapters,
        rescheduled_tasks=len(result.cancelled_task_ids),
        message=message,
    )