pdf-quiz-generator/backend/app/routers/teach.py

"""Teach chat endpoint — AI tutor for study mode questions."""
import logging
from datetime import datetime, time, timedelta, timezone

from fastapi import APIRouter, Depends, HTTPException
from fastapi.concurrency import run_in_threadpool
from pydantic import BaseModel
from sqlalchemy.orm import Session

from app.database import get_db
from app.models.question import Question
from app.models.ai_model_config import AIModelConfig
from app.models.attempt import QuizAttempt
from app.models.user import User
from app.utils.quiz_access import require_question_access
from app.services import question_figures, site_settings, vision_service
from app.services.quiz_builder import bank_question_predicate
from app.utils.auth import check_rate_limit, get_current_user, require_moderator

router = APIRouter()
log = logging.getLogger(__name__)


def _daily_teach_limit() -> tuple[str, int]:
    now = datetime.now(timezone.utc)
    reset_at = datetime.combine(now.date() + timedelta(days=1), time.min, tzinfo=timezone.utc)
    ttl = max(60, int((reset_at - now).total_seconds()) + 300)
    return now.date().isoformat(), ttl


class ChatMessage(BaseModel):
    role: str  # "user" | "assistant"
    content: str


class ChatRequest(BaseModel):
    model_config = {"protected_namespaces": ()}

    question_id: int
    attempt_id: int | None = None
    messages: list[ChatMessage]
    model_id: int | None = None  # AIModelConfig.id — if None, use default


def _get_teach_model(db: Session, model_config_id: int | None = None):
    """Return (model_id, api_key) for the requested (or default) teach model, or None."""
    if model_config_id:
        m = db.query(AIModelConfig).filter(
            AIModelConfig.id == model_config_id,
            AIModelConfig.task == "teach",
            AIModelConfig.is_active == True,
        ).first()
        if m:
            return (m.model_id, m.api_key or None)
    # Fall back to default, then any active
    m = db.query(AIModelConfig).filter(
        AIModelConfig.task == "teach",
        AIModelConfig.is_active == True,
        AIModelConfig.is_default == True,
    ).first()
    if not m:
        m = db.query(AIModelConfig).filter(
            AIModelConfig.task == "teach",
            AIModelConfig.is_active == True,
        ).first()
    return (m.model_id, m.api_key or None) if m else None


def _find_similar_questions(db: Session, question: Question, user: User, limit: int = 4) -> list[Question]:
    """Filter eligible context in SQL before similarity ranking and LIMIT."""
    if question.embedding is None:
        return []
    try:
        distance = Question.embedding.cosine_distance(question.embedding)
        return db.query(Question).filter(
            bank_question_predicate(user), Question.id != question.id,
            Question.embedding.isnot(None), distance <= 0.65,
        ).order_by(distance).limit(limit).all()
    except Exception:
        return []


#: What a figure's role is called when it is described to the tutor. The two
#: are not interchangeable: "the figure in the explanation" and "the figure in
#: the stem" mean different things to something that has been told the answer.
FIGURE_CAPTIONS = {
    "stem": "the figure printed with the question stem",
    "explanation": "the figure printed with the explanation",
}


def _figures(db: Session, question: Question) -> list[vision_service.Image]:
    """The pictures printed with this question, loaded from storage.

    The learner is looking at them. Until now the tutor was not: it was handed
    the stem, the options and the answer key as text and left to teach around a
    radiograph it had never seen, which produces confident sentences about a
    finding nobody described.

    Read from `question_media`, which is where figures live — a question may
    carry several, and the two legacy path columns can hold only one each.
    They agree today, so nothing was being lost yet; the first question given a
    second figure in the editor would have been the one that broke it, silently
    and only for the tutor.
    """
    rows = question_figures.figures_for(db, question.id)
    images = [
        vision_service.load_image(row["path"],
                                  FIGURE_CAPTIONS.get(row.get("role"), "a figure printed with this question"))
        for row in rows if row.get("path")
    ]
    if not images:
        # Nothing projected into `question_media` yet — a question written
        # before that table existed, or one whose backfill has not run.
        legacy = [
            (question.image_path, FIGURE_CAPTIONS["stem"]),
            (question.explanation_image_path, FIGURE_CAPTIONS["explanation"]),
        ]
        images = [vision_service.load_image(path, caption) for path, caption in legacy if path]
    return [image for image in images if image]


def _option_label(index: int) -> str:
    """A, B, C … matching what the learner has on screen.

    The player letters its options. A tutor told they are numbered will say
    "option 3" about the thing the student is looking at as C, and the student
    is then reconciling two labellings of the same five lines.
    """
    return chr(65 + index) if index < 26 else str(index + 1)


def _build_system_prompt(question: Question, similar: list[Question]) -> str:
    correct = (question.correct_answer or "").strip()
    opts = ""
    if question.options:
        # The keyed answer is marked against its own option rather than only
        # quoted underneath. Given the text alone the model has to match a
        # string back to a line, and when two options start the same way it
        # matches the wrong one and teaches from it.
        opts = "\n".join(
            f"  {_option_label(i)}) {opt}" + ("   <-- CORRECT ANSWER" if correct and str(opt).strip() == correct else "")
            for i, opt in enumerate(question.options)
        )

    prompt = (
        "You are a medical education tutor. A student is studying the question below.\n"
        "Rules:\n"
        "- Answer the student's question directly. Do NOT ask clarifying questions.\n"
        "- You may reveal and explain the correct answer and why wrong options are wrong.\n"
        "- Refer to options by their letter, exactly as they are lettered below.\n"
        "- Use markdown formatting: bold key terms, bullet lists for comparisons.\n"
        "- Keep responses under 200 words unless a detailed explanation is needed.\n"
        "- Never ask 'what would you like to know?' — just explain.\n\n"
        f"=== Question ===\n{question.question_text}\n"
    )
    if opts:
        prompt += f"Options:\n{opts}\n"
    if correct:
        prompt += f"Correct Answer: {correct}\n"
    if question.explanation:
        prompt += f"Explanation: {question.explanation}\n"
    if correct:
        # Said as a rule, not as data. Without it a model that would have
        # answered differently argues its own way and tells the student the
        # marked answer is wrong — which is the one thing a tutor sitting
        # beside a marked question must never do.
        prompt += (
            "\nThe answer above is this question's answer key. It is authoritative: "
            "teach the reasoning that leads to it, and never tell the student a "
            "different option is correct. If you believe the key is genuinely "
            "contestable, explain the case for it first and say in one sentence "
            "that the point is debated.\n"
        )

    if similar:
        prompt += "\n=== Related Questions (for broader context) ===\n"
        for i, sq in enumerate(similar, 1):
            prompt += f"{i}. {sq.question_text}"
            if sq.correct_answer:
                prompt += f" → Answer: {sq.correct_answer}"
            prompt += "\n"

    prompt += (
        "\nAnswer the student's question directly. Explain the correct answer, "
        "the underlying concept, and why wrong options are incorrect if relevant. "
        "Do not ask what they want to know — just teach.\n\n"
        "After your explanation, suggest exactly 3 follow-up questions the student might want to ask next. "
        "Put them at the very end, each on its own line starting with '> '. Example:\n"
        "> Why is dopamine not the first-line treatment here?\n"
        "> What are the other causes of this presentation?\n"
        "> How does this differ in neonates vs older children?"
    )
    return prompt


@router.get("/models")
def list_teach_models(
    db: Session = Depends(get_db),
    current_user: User = Depends(get_current_user),
):
    """Return available teach models for the frontend to display."""
    models = db.query(AIModelConfig).filter(
        AIModelConfig.task == "teach",
        AIModelConfig.is_active == True,
    ).all()
    return [{"id": m.id, "name": m.name, "model_id": m.model_id, "is_default": m.is_default} for m in models]


@router.get("/policy")
def tutor_policy(current_user: User = Depends(get_current_user)):
    """Whether the tutor may be opened during a session.

    Asked before the quiz offers it, so a switch an administrator has thrown
    reads as "not there" rather than as a button that fails when pressed.
    Study mode is a separate rule the interface already applies and the server
    enforces regardless of this answer.
    """
    return {"in_quiz": site_settings.get_flag("tutor_in_quiz")}


@router.get("/prompt")
def show_prompt(
    db: Session = Depends(get_db),
    current_user: User = Depends(require_moderator),
):
    """What the tutor is actually told, for anyone who maintains questions.

    An educator answering "why did the tutor say that?" should be able to read
    the instructions rather than infer them, and the answer matters: the tutor
    is handed the correct answer and the explanation, and is told it may
    reveal them. That is the reason it is never offered during a running exam.

    Rendered against a stand-in question so the shape is visible without
    naming any real one.
    """
    example = Question(
        id=0,
        question_text="<the question the learner is on>",
        options=["<option 1>", "<option 2>"],
        correct_answer="<the correct option>",
        explanation="<the explanation stored on the question>",
        question_type="mcq",
    )
    return {
        "prompt": _build_system_prompt(example, []),
        "sends_correct_answer": True,
        "sends_explanation": True,
        "similar_questions": 4,
        "daily_limit": 30,
        "notes": [
            "The tutor is given the correct answer and the explanation, and is"
            " told it may reveal them. It is therefore never offered while an"
            " exam-mode attempt is still running.",
            "Up to four semantically similar questions are added for context,"
            " with their answers. They are chosen by embedding, not by hand.",
            "An administrator can turn the tutor off for study sessions too,"
            " under Settings, Access and joining.",
            "Editing this text means changing _build_system_prompt in"
            " backend/app/routers/teach.py — it is code, not a setting, so a"
            " change is reviewed and deployed like any other.",
        ],
    }


def _require_tutor_allowed(db: Session, attempt_id: int | None, user: User) -> None:
    """Refuse the tutor mid-session when an administrator has turned it off.

    Enforced here rather than only in the interface, because hiding a button
    does not stop a request. Two rules, and the first is not a setting:

    * Never during an exam that is still running. The prompt below is given
      the correct answer and told it may explain it, so an exam-mode tutor is
      an answer key. `require_question_access` already refuses this.
    * Optionally not during a study session either, which is what this flag
      decides. Reviewing a finished attempt is not "during", so it is
      unaffected — the answers are already shown by then.
    """
    if attempt_id is None:
        return
    attempt = db.query(QuizAttempt).filter_by(id=attempt_id, user_id=user.id).first()
    if attempt is None or attempt.completed_at is not None:
        return
    if not site_settings.get_flag("tutor_in_quiz"):
        raise HTTPException(403, "The tutor is turned off while a session is being sat")


@router.post("/chat")
async def chat(
    req: ChatRequest,
    db: Session = Depends(get_db),
    current_user: User = Depends(get_current_user),
):
    """Send a message to the teach AI with full question context."""
    question = db.query(Question).filter(Question.id == req.question_id).first()
    require_question_access(db, question, current_user, req.attempt_id, review=True)
    _require_tutor_allowed(db, req.attempt_id, current_user)

    # Daily AI coach quota. Admins, moderators, and unthrottled users are exempt.
    quota_day, quota_ttl = _daily_teach_limit()
    check_rate_limit(
        key=f"teach_chat_daily:{current_user.id}:{quota_day}",
        max_calls=30,
        window_seconds=quota_ttl,
        detail="You've reached today's AI coach limit of 30 messages. Try again tomorrow, or contact an admin if you need the limit raised.",
        user=current_user,
    )
    model_info = _get_teach_model(db, req.model_id)
    if not model_info:
        raise HTTPException(
            status_code=503,
            detail="No teaching AI model is configured. Ask an admin to add a model with task 'teach'.",
        )
    model_id, api_key = model_info

    similar = _find_similar_questions(db, question, current_user)
    system_prompt = _build_system_prompt(question, similar)

    messages = [{"role": "system", "content": system_prompt}]

    # Blocking work — object storage, and possibly a second model — inside an
    # async endpoint, so it goes to a worker thread rather than stopping every
    # other request on this uvicorn worker.
    handoff = None
    figures = await run_in_threadpool(_figures, db, question)
    if figures:
        try:
            parts, handoff = await run_in_threadpool(
                vision_service.image_context, db, figures,
                model_id=model_id, api_key=api_key,
                context=(question.question_text or "")[:1500])
        except vision_service.VisionUnavailable as e:
            # The figure is context this endpoint volunteered, not the learner's
            # question, so nothing being able to read it costs the turn its
            # picture rather than ending it. The tutor is told in the prompt
            # all the same: a tutor that does not know it is blind writes a
            # paragraph about a radiograph nobody looked at.
            log.warning("Tutor figure went unread for question %s: %s", question.id, e)
            parts = [{"type": "text", "text": (
                "This question is printed with a figure. Nothing available can read it "
                "for you — an administrator has not configured a Tool model. Teach from "
                "the text, and neither describe nor infer what the figure shows.")}]
        messages.append({"role": "user", "content": parts})

    for msg in req.messages:
        if msg.role not in ("user", "assistant"):
            continue
        messages.append({"role": msg.role, "content": msg.content})

    try:
        from app.services.ai_service import DEFAULT_TIMEOUT, achat
        raw = (await achat(
            model=model_id,
            messages=messages,
            max_tokens=600,
            temperature=0.4,
            api_key=api_key,
            timeout=DEFAULT_TIMEOUT,
        )).strip()

        # Parse out follow-up suggestions (lines starting with "> ")
        lines = raw.splitlines()
        suggestions = [l[2:].strip() for l in lines if l.startswith("> ")]
        reply_lines = [l for l in lines if not l.startswith("> ")]
        # Trim trailing blank lines from reply
        while reply_lines and not reply_lines[-1].strip():
            reply_lines.pop()
        reply = "\n".join(reply_lines).strip()

        answer = {"reply": reply, "suggestions": suggestions[:3]}
        if handoff and handoff.delegated:
            # Two models ran. Said in the response as well as the log, so that
            # "why was that turn slow" has an answer on the page it happened on.
            answer["vision"] = handoff.as_dict()
        return answer
    except Exception as e:
        log.error(f"TeachChat error for user {current_user.id} model {model_id}: {e}")
        raise HTTPException(status_code=502, detail="The AI tutor is temporarily unavailable. Please try again in a moment.")