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