refactor(quiz): convert generate_quiz to quiz_agent (refactor #2)

backend/agents/_providers.py

@@ -7,6 +7,7 @@
  SAPLING_MODEL_SUMMARY=gemini-2.5-flash-lite
  SAPLING_MODEL_CONCEPTS=gemini-2.5-flash
  SAPLING_MODEL_SYLLABUS=gemini-2.5-flash
+ SAPLING_MODEL_QUIZ=gemini-2.5-flash-lite
 
 Defaults are tuned per task: cheaper models for simpler classifications,
 flagship Flash for tasks where output quality drives downstream UX.
@@ -23,7 +24,7 @@
 from config import GEMINI_API_KEY
 
 
-AgentTask = Literal["classifier", "summary", "concepts", "syllabus"]
+AgentTask = Literal["classifier", "summary", "concepts", "syllabus", "quiz"]
 
 
 # Defaults are conservative. Bumping a model up costs more; the env var
@@ -33,6 +34,10 @@
  "summary": "gemini-2.5-flash-lite",
  "concepts": "gemini-2.5-flash",
  "syllabus": "gemini-2.5-flash",
+ # Quiz generation defaults to lite: it's a single-shot non-streaming
+ # call where the agent pulls structured graph data via tools, so the
+ # bulk of the value is in tool wiring, not raw model strength.
+ "quiz": "gemini-2.5-flash-lite",
 }
 
 

backend/agents/quiz.py

@@ -0,0 +1,111 @@
+"""Quiz-generation agent.
+
+Replaces routes/quiz.py:82's call_gemini_json + manual prompt-string
+augmentation. The agent has tools to pull weak concepts + class
+misconceptions on demand instead of pre-stuffing them into the prompt.
+
+Per ADR 0003 convention 4: keep the output schema compact. Gemini's
+structured-output API rejects rich nested schemas with too many states
+for serving — Quiz is a flat top-level model with a list of
+QuizQuestion items, no further nesting.
+"""
+
+from __future__ import annotations
+
+import hashlib
+from typing import Literal
+
+from pydantic import BaseModel, Field
+from pydantic_ai import Agent
+
+from agents._providers import model_for
+from agents.deps import SaplingDeps
+from agents.tools.graph_read import (
+ read_concepts_for_user_tool,
+ read_misconceptions_for_course_tool,
+)
+
+
+# Difficulty + question type are Literals so Gemini's enum constraint
+# applies and downstream UI can branch on stable strings.
+#
+# `QuizQuestionType` is intentionally MCQ-only today. The frontend
+# `submitQuiz` flow grades by `q["options"][i].correct` lookup — there's
+# no UI for free-text answers, no fuzzy-match grading, no LLM-judged
+# scoring. Generating short-answer questions through this path would
+# emit unrenderable, ungradable items. Keep the type narrow until real
+# short-answer support exists; revisit when that lands.
+QuizDifficulty = Literal["easy", "medium", "hard"]
+QuizQuestionType = Literal["multiple_choice"]
+
+
+class QuizQuestion(BaseModel):
+ """A single multiple-choice quiz question. Kept small so the parent
+ Quiz schema doesn't trip Gemini's structured-output complexity limit."""
+
+ question: str = Field(max_length=600)
+ type: QuizQuestionType
+ difficulty: QuizDifficulty
+ # 3-6 options. The agent is required to populate this for every
+ # question and to make `correct_answer` match exactly one of them.
+ options: list[str] = Field(min_length=3, max_length=6)
+ # The option text the agent considers correct. Must appear verbatim
+ # in `options`; the route validates this and drops questions that
+ # violate the contract rather than silently mis-marking them.
+ correct_answer: str = Field(max_length=400)
+ explanation: str = Field(max_length=600)
+ # Concept the question is testing — must be one of the user's known
+ # concept_names per the prompt. Used by the route to award mastery
+ # on a correct answer.
+ concept: str = Field(max_length=120)
+
+
+class Quiz(BaseModel):
+ """The agent's structured output."""
+
+ questions: list[QuizQuestion] = Field(min_length=1, max_length=20)
+
+
+_SYSTEM_PROMPT = (
+ "You generate adaptive multiple-choice quizzes for a student. Each "
+ "question must target a specific concept the student has weak "
+ "mastery on, OR address a class-level misconception you've seen.\n\n"
+ "Workflow:\n"
+ "1. Call `read_concepts_for_user` to see the student's mastery per "
+ " concept for this course (returned sorted by mastery ASC — "
+ " weakest first).\n"
+ "2. Call `read_misconceptions_for_course` to see anonymized class "
+ " misconceptions. Use these to phrase distractors and to write "
+ " a question that probes the misconception.\n"
+ "3. Compose `Quiz.questions` so the WEAKEST concepts get the most "
+ " questions, AND each item's `concept` field exactly matches a "
+ " concept_name returned by tool 1.\n\n"
+ "Per-question rules (multiple-choice only — the type field is "
+ "constrained to 'multiple_choice'):\n"
+ "- 4 options, exactly one correct. The text in `correct_answer` "
+ " MUST appear verbatim in `options` — character-for-character. "
+ " Questions that violate this are dropped at the route layer.\n"
+ "- Distractors should reflect plausible misconceptions, not random "
+ " noise. Use the read_misconceptions_for_course return value.\n"
+ "- explanation: 1-3 sentences explaining WHY the correct answer "
+ " is correct — used in the post-quiz review screen.\n"
+ "- difficulty: align with the student's mastery on the concept; "
+ " weakest concepts get easy/medium, strongest get hard.\n\n"
+ "Honor the requested num_questions and difficulty distribution "
+ "in the user message. Don't invent concepts the student doesn't "
+ "have."
+)
+_PROMPT_HASH = hashlib.sha256(_SYSTEM_PROMPT.encode("utf-8")).hexdigest()[:12]
+
+
+quiz_agent = Agent[SaplingDeps, Quiz](
+ model=model_for("quiz"),
+ deps_type=SaplingDeps,
+ output_type=Quiz,
+ system_prompt=_SYSTEM_PROMPT,
+ metadata={"prompt_version": _PROMPT_HASH, "agent": "quiz"},
+ tools=[
+ read_concepts_for_user_tool,
+ read_misconceptions_for_course_tool,
+ ],
+)

backend/agents/tools/graph_read.py

@@ -0,0 +1,163 @@
+"""Read-side graph tools for Pydantic AI agents.
+
+Per ADR 0004: the agent needs to pull mastery state + course-level
+misconceptions when planning a quiz. These are the tool surfaces.
+The pure-async functions are callable directly from routes;
+the *_tool wrappers register on a Pydantic AI Agent.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+from typing import Any
+
+from pydantic import BaseModel, Field
+from pydantic_ai import RunContext
+
+from agents.deps import SaplingDeps
+from db.connection import table
+
+logger = logging.getLogger(__name__)
+
+
+# ── read_concepts_for_user ────────────────────────────────────────────────
+
+
+class ConceptMastery(BaseModel):
+ """Per-concept mastery state for the user in a course."""
+
+ concept_name: str
+ mastery: float = Field(ge=0.0, le=1.0)
+ last_reviewed_at: str | None = None
+
+
+async def read_concepts_for_user(
+ user_id: str,
+ course_id: str | None,
+) -> list[ConceptMastery]:
+ """Return the user's concept mastery for a course (or globally if
+ course_id is None). Sorted by mastery ASC so the weakest concepts
+ appear first — quiz_agent uses this ordering to focus on weak areas.
+
+ Pure async, callable from routes. Wraps the underlying sync
+ Supabase read in asyncio.to_thread so it doesn't block the loop.
+
+ NOTE: The underlying `graph_nodes` table stores the mastery value as
+ `mastery_score` and the timestamp as `last_studied_at`. We map them
+ here to the agent-facing names (`mastery`, `last_reviewed_at`) so the
+ tool contract stays stable even if storage column names change.
+ """
+
+ def _fetch() -> list[dict[str, Any]]:
+ filters = {"user_id": f"eq.{user_id}"}
+ if course_id:
+ filters["course_id"] = f"eq.{course_id}"
+ try:
+ return (
+ table("graph_nodes").select(
+ "concept_name,mastery_score,last_studied_at",
+ filters=filters,
+ order="mastery_score.asc",
+ )
+ or []
+ )
+ except Exception:
+ logger.exception(
+ "read_concepts_for_user failed for user=%s course=%s",
+ user_id,
+ course_id,
+ )
+ return []
+
+ rows = await asyncio.to_thread(_fetch)
+ return [
+ ConceptMastery(
+ concept_name=r.get("concept_name") or "",
+ mastery=float(r.get("mastery_score") or 0.0),
+ last_reviewed_at=r.get("last_studied_at"),
+ )
+ for r in rows
+ if r.get("concept_name")
+ ]
+
+
+async def read_concepts_for_user_tool(
+ ctx: RunContext[SaplingDeps],
+) -> list[ConceptMastery]:
+ """Pydantic AI tool wrapper. Reads from ctx.deps."""
+ return await read_concepts_for_user(ctx.deps.user_id, ctx.deps.course_id)
+
+
+# ── read_misconceptions_for_course ────────────────────────────────────────
+
+
+class Misconception(BaseModel):
+ """A class-level misconception observed across student sessions."""
+
+ text: str
+ related_concept: str | None = None
+
+
+async def read_misconceptions_for_course(
+ course_id: str | None,
+) -> list[Misconception]:
+ """Return aggregated misconception strings for a course. Anonymized
+ (sourced from class-wide patterns, not from any single student).
+ Returns [] when course_id is None or the underlying table is empty.
+
+ Source: `course_concept_stats` rows for the course. Each row
+ represents one concept and carries a `common_misconceptions` array
+ (populated by the hash-gated aggregation in
+ `services/course_context_service.py`). We flatten each array entry
+ into its own Misconception, tagging `related_concept` with the
+ concept name so the agent can route distractors per-concept.
+
+ The spec referenced a hypothetical `misconceptions` table — that
+ table does not exist in this schema. `course_concept_stats` is the
+ real source of class-wide misconception strings, so we read that.
+ The tool contract (returning Misconception[]) is unchanged.
+ """
+ if not course_id:
+ return []
+
+ def _fetch() -> list[dict[str, Any]]:
+ try:
+ return (
+ table("course_concept_stats").select(
+ "concept_name,common_misconceptions",
+ filters={"course_id": f"eq.{course_id}"},
+ order="updated_at.desc",
+ limit=20,
+ )
+ or []
+ )
+ except Exception:
+ logger.exception(
+ "read_misconceptions_for_course failed for course=%s",
+ course_id,
+ )
+ return []
+
+ rows = await asyncio.to_thread(_fetch)
+ out: list[Misconception] = []
+ seen: set[str] = set()
+ for r in rows:
+ concept = r.get("concept_name") or None
+ for m in r.get("common_misconceptions") or []:
+ text = (m or "").strip() if isinstance(m, str) else ""
+ if not text:
+ continue
+ key = text.lower()
+ if key in seen:
+ continue
+ seen.add(key)
+ out.append(Misconception(text=text, related_concept=concept))
+ return out
+
+
+async def read_misconceptions_for_course_tool(
+ ctx: RunContext[SaplingDeps],
+) -> list[Misconception]:
+ """Pydantic AI tool wrapper. Reads from ctx.deps."""
+ return await read_misconceptions_for_course(ctx.deps.course_id)

backend/models/__init__.py

@@ -44,6 +44,12 @@ class GenerateQuizBody(BaseModel):
  num_questions: int = 5
  difficulty: str = "medium"
  use_shared_context: bool = True
+ # Mirrors the Learn-route fast/smart toggle so quiz generation has
+ # the same per-request model-quality choice as chat tutoring.
+ # "fast" → gemini-2.5-flash, "smart" → gemini-2.5-pro. None falls
+ # through to whatever SAPLING_MODEL_QUIZ resolves to (default
+ # gemini-2.5-flash-lite per ADR 0008).
+ model_pref: Optional[Literal["fast", "smart"]] = None
 
 
 class AnswerItem(BaseModel):

backend/agents/_providers.py

@@ -7,6 +7,7 @@
  SAPLING_MODEL_SUMMARY=gemini-2.5-flash-lite
  SAPLING_MODEL_CONCEPTS=gemini-2.5-flash
  SAPLING_MODEL_SYLLABUS=gemini-2.5-flash
+ SAPLING_MODEL_QUIZ=gemini-2.5-flash-lite
 
 Defaults are tuned per task: cheaper models for simpler classifications,
 flagship Flash for tasks where output quality drives downstream UX.
@@ -23,7 +24,7 @@
 from config import GEMINI_API_KEY
 
 
-AgentTask = Literal["classifier", "summary", "concepts", "syllabus"]
+AgentTask = Literal["classifier", "summary", "concepts", "syllabus", "quiz"]
 
 
 # Defaults are conservative. Bumping a model up costs more; the env var
@@ -33,6 +34,10 @@
  "summary": "gemini-2.5-flash-lite",
  "concepts": "gemini-2.5-flash",
  "syllabus": "gemini-2.5-flash",
+ # Quiz generation defaults to lite: it's a single-shot non-streaming
+ # call where the agent pulls structured graph data via tools, so the
+ # bulk of the value is in tool wiring, not raw model strength.
+ "quiz": "gemini-2.5-flash-lite",
 }
 
 

backend/agents/quiz.py

@@ -0,0 +1,111 @@
+"""Quiz-generation agent.
+
+Replaces routes/quiz.py:82's call_gemini_json + manual prompt-string
+augmentation. The agent has tools to pull weak concepts + class
+misconceptions on demand instead of pre-stuffing them into the prompt.
+
+Per ADR 0003 convention 4: keep the output schema compact. Gemini's
+structured-output API rejects rich nested schemas with too many states
+for serving — Quiz is a flat top-level model with a list of
+QuizQuestion items, no further nesting.
+"""
+
+from __future__ import annotations
+
+import hashlib
+from typing import Literal
+
+from pydantic import BaseModel, Field
+from pydantic_ai import Agent
+
+from agents._providers import model_for
+from agents.deps import SaplingDeps
+from agents.tools.graph_read import (
+ read_concepts_for_user_tool,
+ read_misconceptions_for_course_tool,
+)
+
+
+# Difficulty + question type are Literals so Gemini's enum constraint
+# applies and downstream UI can branch on stable strings.
+#
+# `QuizQuestionType` is intentionally MCQ-only today. The frontend
+# `submitQuiz` flow grades by `q["options"][i].correct` lookup — there's
+# no UI for free-text answers, no fuzzy-match grading, no LLM-judged
+# scoring. Generating short-answer questions through this path would
+# emit unrenderable, ungradable items. Keep the type narrow until real
+# short-answer support exists; revisit when that lands.
+QuizDifficulty = Literal["easy", "medium", "hard"]
+QuizQuestionType = Literal["multiple_choice"]
+
+
+class QuizQuestion(BaseModel):
+ """A single multiple-choice quiz question. Kept small so the parent
+ Quiz schema doesn't trip Gemini's structured-output complexity limit."""
+
+ question: str = Field(max_length=600)
+ type: QuizQuestionType
+ difficulty: QuizDifficulty
+ # 3-6 options. The agent is required to populate this for every
+ # question and to make `correct_answer` match exactly one of them.
+ options: list[str] = Field(min_length=3, max_length=6)
+ # The option text the agent considers correct. Must appear verbatim
+ # in `options`; the route validates this and drops questions that
+ # violate the contract rather than silently mis-marking them.
+ correct_answer: str = Field(max_length=400)
+ explanation: str = Field(max_length=600)
+ # Concept the question is testing — must be one of the user's known
+ # concept_names per the prompt. Used by the route to award mastery
+ # on a correct answer.
+ concept: str = Field(max_length=120)
+
+
+class Quiz(BaseModel):
+ """The agent's structured output."""
+
+ questions: list[QuizQuestion] = Field(min_length=1, max_length=20)
+
+
+_SYSTEM_PROMPT = (
+ "You generate adaptive multiple-choice quizzes for a student. Each "
+ "question must target a specific concept the student has weak "
+ "mastery on, OR address a class-level misconception you've seen.\n\n"
+ "Workflow:\n"
+ "1. Call `read_concepts_for_user` to see the student's mastery per "
+ " concept for this course (returned sorted by mastery ASC — "
+ " weakest first).\n"
+ "2. Call `read_misconceptions_for_course` to see anonymized class "
+ " misconceptions. Use these to phrase distractors and to write "
+ " a question that probes the misconception.\n"
+ "3. Compose `Quiz.questions` so the WEAKEST concepts get the most "
+ " questions, AND each item's `concept` field exactly matches a "
+ " concept_name returned by tool 1.\n\n"
+ "Per-question rules (multiple-choice only — the type field is "
+ "constrained to 'multiple_choice'):\n"
+ "- 4 options, exactly one correct. The text in `correct_answer` "
+ " MUST appear verbatim in `options` — character-for-character. "
+ " Questions that violate this are dropped at the route layer.\n"
+ "- Distractors should reflect plausible misconceptions, not random "
+ " noise. Use the read_misconceptions_for_course return value.\n"
+ "- explanation: 1-3 sentences explaining WHY the correct answer "
+ " is correct — used in the post-quiz review screen.\n"
+ "- difficulty: align with the student's mastery on the concept; "
+ " weakest concepts get easy/medium, strongest get hard.\n\n"
+ "Honor the requested num_questions and difficulty distribution "
+ "in the user message. Don't invent concepts the student doesn't "
+ "have."
+)
+_PROMPT_HASH = hashlib.sha256(_SYSTEM_PROMPT.encode("utf-8")).hexdigest()[:12]
+
+
+quiz_agent = Agent[SaplingDeps, Quiz](
+ model=model_for("quiz"),
+ deps_type=SaplingDeps,
+ output_type=Quiz,
+ system_prompt=_SYSTEM_PROMPT,
+ metadata={"prompt_version": _PROMPT_HASH, "agent": "quiz"},
+ tools=[
+ read_concepts_for_user_tool,
+ read_misconceptions_for_course_tool,
+ ],
+)

backend/agents/tools/graph_read.py

@@ -0,0 +1,163 @@
+"""Read-side graph tools for Pydantic AI agents.
+
+Per ADR 0004: the agent needs to pull mastery state + course-level
+misconceptions when planning a quiz. These are the tool surfaces.
+The pure-async functions are callable directly from routes;
+the *_tool wrappers register on a Pydantic AI Agent.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+from typing import Any
+
+from pydantic import BaseModel, Field
+from pydantic_ai import RunContext
+
+from agents.deps import SaplingDeps
+from db.connection import table
+
+logger = logging.getLogger(__name__)
+
+
+# ── read_concepts_for_user ────────────────────────────────────────────────
+
+
+class ConceptMastery(BaseModel):
+ """Per-concept mastery state for the user in a course."""
+
+ concept_name: str
+ mastery: float = Field(ge=0.0, le=1.0)
+ last_reviewed_at: str | None = None
+
+
+async def read_concepts_for_user(
+ user_id: str,
+ course_id: str | None,
+) -> list[ConceptMastery]:
+ """Return the user's concept mastery for a course (or globally if
+ course_id is None). Sorted by mastery ASC so the weakest concepts
+ appear first — quiz_agent uses this ordering to focus on weak areas.
+
+ Pure async, callable from routes. Wraps the underlying sync
+ Supabase read in asyncio.to_thread so it doesn't block the loop.
+
+ NOTE: The underlying `graph_nodes` table stores the mastery value as
+ `mastery_score` and the timestamp as `last_studied_at`. We map them
+ here to the agent-facing names (`mastery`, `last_reviewed_at`) so the
+ tool contract stays stable even if storage column names change.
+ """
+
+ def _fetch() -> list[dict[str, Any]]:
+ filters = {"user_id": f"eq.{user_id}"}
+ if course_id:
+ filters["course_id"] = f"eq.{course_id}"
+ try:
+ return (
+ table("graph_nodes").select(
+ "concept_name,mastery_score,last_studied_at",
+ filters=filters,
+ order="mastery_score.asc",
+ )
+ or []
+ )
+ except Exception:
+ logger.exception(
+ "read_concepts_for_user failed for user=%s course=%s",
+ user_id,
+ course_id,
+ )
+ return []
+
+ rows = await asyncio.to_thread(_fetch)
+ return [
+ ConceptMastery(
+ concept_name=r.get("concept_name") or "",
+ mastery=float(r.get("mastery_score") or 0.0),
+ last_reviewed_at=r.get("last_studied_at"),
+ )
+ for r in rows
+ if r.get("concept_name")
+ ]
+
+
+async def read_concepts_for_user_tool(
+ ctx: RunContext[SaplingDeps],
+) -> list[ConceptMastery]:
+ """Pydantic AI tool wrapper. Reads from ctx.deps."""
+ return await read_concepts_for_user(ctx.deps.user_id, ctx.deps.course_id)
+
+
+# ── read_misconceptions_for_course ────────────────────────────────────────
+
+
+class Misconception(BaseModel):
+ """A class-level misconception observed across student sessions."""
+
+ text: str
+ related_concept: str | None = None
+
+
+async def read_misconceptions_for_course(
+ course_id: str | None,
+) -> list[Misconception]:
+ """Return aggregated misconception strings for a course. Anonymized
+ (sourced from class-wide patterns, not from any single student).
+ Returns [] when course_id is None or the underlying table is empty.
+
+ Source: `course_concept_stats` rows for the course. Each row
+ represents one concept and carries a `common_misconceptions` array
+ (populated by the hash-gated aggregation in
+ `services/course_context_service.py`). We flatten each array entry
+ into its own Misconception, tagging `related_concept` with the
+ concept name so the agent can route distractors per-concept.
+
+ The spec referenced a hypothetical `misconceptions` table — that
+ table does not exist in this schema. `course_concept_stats` is the
+ real source of class-wide misconception strings, so we read that.
+ The tool contract (returning Misconception[]) is unchanged.
+ """
+ if not course_id:
+ return []
+
+ def _fetch() -> list[dict[str, Any]]:
+ try:
+ return (
+ table("course_concept_stats").select(
+ "concept_name,common_misconceptions",
+ filters={"course_id": f"eq.{course_id}"},
+ order="updated_at.desc",
+ limit=20,
+ )
+ or []
+ )
+ except Exception:
+ logger.exception(
+ "read_misconceptions_for_course failed for course=%s",
+ course_id,
+ )
+ return []
+
+ rows = await asyncio.to_thread(_fetch)
+ out: list[Misconception] = []
+ seen: set[str] = set()
+ for r in rows:
+ concept = r.get("concept_name") or None
+ for m in r.get("common_misconceptions") or []:
+ text = (m or "").strip() if isinstance(m, str) else ""
+ if not text:
+ continue
+ key = text.lower()
+ if key in seen:
+ continue
+ seen.add(key)
+ out.append(Misconception(text=text, related_concept=concept))
+ return out
+
+
+async def read_misconceptions_for_course_tool(
+ ctx: RunContext[SaplingDeps],
+) -> list[Misconception]:
+ """Pydantic AI tool wrapper. Reads from ctx.deps."""
+ return await read_misconceptions_for_course(ctx.deps.course_id)

backend/models/__init__.py

@@ -44,6 +44,12 @@ class GenerateQuizBody(BaseModel):
  num_questions: int = 5
  difficulty: str = "medium"
  use_shared_context: bool = True
+ # Mirrors the Learn-route fast/smart toggle so quiz generation has
+ # the same per-request model-quality choice as chat tutoring.
+ # "fast" → gemini-2.5-flash, "smart" → gemini-2.5-pro. None falls
+ # through to whatever SAPLING_MODEL_QUIZ resolves to (default
+ # gemini-2.5-flash-lite per ADR 0008).
+ model_pref: Optional[Literal["fast", "smart"]] = None
 
 
 class AnswerItem(BaseModel):

backend/agents/_providers.py

@@ -7,6 +7,7 @@
  SAPLING_MODEL_SUMMARY=gemini-2.5-flash-lite
  SAPLING_MODEL_CONCEPTS=gemini-2.5-flash
  SAPLING_MODEL_SYLLABUS=gemini-2.5-flash
+ SAPLING_MODEL_QUIZ=gemini-2.5-flash-lite
 
 Defaults are tuned per task: cheaper models for simpler classifications,
 flagship Flash for tasks where output quality drives downstream UX.
@@ -23,7 +24,7 @@
 from config import GEMINI_API_KEY
 
 
-AgentTask = Literal["classifier", "summary", "concepts", "syllabus"]
+AgentTask = Literal["classifier", "summary", "concepts", "syllabus", "quiz"]
 
 
 # Defaults are conservative. Bumping a model up costs more; the env var
@@ -33,6 +34,10 @@
  "summary": "gemini-2.5-flash-lite",
  "concepts": "gemini-2.5-flash",
  "syllabus": "gemini-2.5-flash",
+ # Quiz generation defaults to lite: it's a single-shot non-streaming
+ # call where the agent pulls structured graph data via tools, so the
+ # bulk of the value is in tool wiring, not raw model strength.
+ "quiz": "gemini-2.5-flash-lite",
 }
 
 

backend/agents/quiz.py

@@ -0,0 +1,111 @@
+"""Quiz-generation agent.
+
+Replaces routes/quiz.py:82's call_gemini_json + manual prompt-string
+augmentation. The agent has tools to pull weak concepts + class
+misconceptions on demand instead of pre-stuffing them into the prompt.
+
+Per ADR 0003 convention 4: keep the output schema compact. Gemini's
+structured-output API rejects rich nested schemas with too many states
+for serving — Quiz is a flat top-level model with a list of
+QuizQuestion items, no further nesting.
+"""
+
+from __future__ import annotations
+
+import hashlib
+from typing import Literal
+
+from pydantic import BaseModel, Field
+from pydantic_ai import Agent
+
+from agents._providers import model_for
+from agents.deps import SaplingDeps
+from agents.tools.graph_read import (
+ read_concepts_for_user_tool,
+ read_misconceptions_for_course_tool,
+)
+
+
+# Difficulty + question type are Literals so Gemini's enum constraint
+# applies and downstream UI can branch on stable strings.
+#
+# `QuizQuestionType` is intentionally MCQ-only today. The frontend
+# `submitQuiz` flow grades by `q["options"][i].correct` lookup — there's
+# no UI for free-text answers, no fuzzy-match grading, no LLM-judged
+# scoring. Generating short-answer questions through this path would
+# emit unrenderable, ungradable items. Keep the type narrow until real
+# short-answer support exists; revisit when that lands.
+QuizDifficulty = Literal["easy", "medium", "hard"]
+QuizQuestionType = Literal["multiple_choice"]
+
+
+class QuizQuestion(BaseModel):
+ """A single multiple-choice quiz question. Kept small so the parent
+ Quiz schema doesn't trip Gemini's structured-output complexity limit."""
+
+ question: str = Field(max_length=600)
+ type: QuizQuestionType
+ difficulty: QuizDifficulty
+ # 3-6 options. The agent is required to populate this for every
+ # question and to make `correct_answer` match exactly one of them.
+ options: list[str] = Field(min_length=3, max_length=6)
+ # The option text the agent considers correct. Must appear verbatim
+ # in `options`; the route validates this and drops questions that
+ # violate the contract rather than silently mis-marking them.
+ correct_answer: str = Field(max_length=400)
+ explanation: str = Field(max_length=600)
+ # Concept the question is testing — must be one of the user's known
+ # concept_names per the prompt. Used by the route to award mastery
+ # on a correct answer.
+ concept: str = Field(max_length=120)
+
+
+class Quiz(BaseModel):
+ """The agent's structured output."""
+
+ questions: list[QuizQuestion] = Field(min_length=1, max_length=20)
+
+
+_SYSTEM_PROMPT = (
+ "You generate adaptive multiple-choice quizzes for a student. Each "
+ "question must target a specific concept the student has weak "
+ "mastery on, OR address a class-level misconception you've seen.\n\n"
+ "Workflow:\n"
+ "1. Call `read_concepts_for_user` to see the student's mastery per "
+ " concept for this course (returned sorted by mastery ASC — "
+ " weakest first).\n"
+ "2. Call `read_misconceptions_for_course` to see anonymized class "
+ " misconceptions. Use these to phrase distractors and to write "
+ " a question that probes the misconception.\n"
+ "3. Compose `Quiz.questions` so the WEAKEST concepts get the most "
+ " questions, AND each item's `concept` field exactly matches a "
+ " concept_name returned by tool 1.\n\n"
+ "Per-question rules (multiple-choice only — the type field is "
+ "constrained to 'multiple_choice'):\n"
+ "- 4 options, exactly one correct. The text in `correct_answer` "
+ " MUST appear verbatim in `options` — character-for-character. "
+ " Questions that violate this are dropped at the route layer.\n"
+ "- Distractors should reflect plausible misconceptions, not random "
+ " noise. Use the read_misconceptions_for_course return value.\n"
+ "- explanation: 1-3 sentences explaining WHY the correct answer "
+ " is correct — used in the post-quiz review screen.\n"
+ "- difficulty: align with the student's mastery on the concept; "
+ " weakest concepts get easy/medium, strongest get hard.\n\n"
+ "Honor the requested num_questions and difficulty distribution "
+ "in the user message. Don't invent concepts the student doesn't "
+ "have."
+)
+_PROMPT_HASH = hashlib.sha256(_SYSTEM_PROMPT.encode("utf-8")).hexdigest()[:12]
+
+
+quiz_agent = Agent[SaplingDeps, Quiz](
+ model=model_for("quiz"),
+ deps_type=SaplingDeps,
+ output_type=Quiz,
+ system_prompt=_SYSTEM_PROMPT,
+ metadata={"prompt_version": _PROMPT_HASH, "agent": "quiz"},
+ tools=[
+ read_concepts_for_user_tool,
+ read_misconceptions_for_course_tool,
+ ],
+)

backend/agents/tools/graph_read.py

@@ -0,0 +1,163 @@
+"""Read-side graph tools for Pydantic AI agents.
+
+Per ADR 0004: the agent needs to pull mastery state + course-level
+misconceptions when planning a quiz. These are the tool surfaces.
+The pure-async functions are callable directly from routes;
+the *_tool wrappers register on a Pydantic AI Agent.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+from typing import Any
+
+from pydantic import BaseModel, Field
+from pydantic_ai import RunContext
+
+from agents.deps import SaplingDeps
+from db.connection import table
+
+logger = logging.getLogger(__name__)
+
+
+# ── read_concepts_for_user ────────────────────────────────────────────────
+
+
+class ConceptMastery(BaseModel):
+ """Per-concept mastery state for the user in a course."""
+
+ concept_name: str
+ mastery: float = Field(ge=0.0, le=1.0)
+ last_reviewed_at: str | None = None
+
+
+async def read_concepts_for_user(
+ user_id: str,
+ course_id: str | None,
+) -> list[ConceptMastery]:
+ """Return the user's concept mastery for a course (or globally if
+ course_id is None). Sorted by mastery ASC so the weakest concepts
+ appear first — quiz_agent uses this ordering to focus on weak areas.
+
+ Pure async, callable from routes. Wraps the underlying sync
+ Supabase read in asyncio.to_thread so it doesn't block the loop.
+
+ NOTE: The underlying `graph_nodes` table stores the mastery value as
+ `mastery_score` and the timestamp as `last_studied_at`. We map them
+ here to the agent-facing names (`mastery`, `last_reviewed_at`) so the
+ tool contract stays stable even if storage column names change.
+ """
+
+ def _fetch() -> list[dict[str, Any]]:
+ filters = {"user_id": f"eq.{user_id}"}
+ if course_id:
+ filters["course_id"] = f"eq.{course_id}"
+ try:
+ return (
+ table("graph_nodes").select(
+ "concept_name,mastery_score,last_studied_at",
+ filters=filters,
+ order="mastery_score.asc",
+ )
+ or []
+ )
+ except Exception:
+ logger.exception(
+ "read_concepts_for_user failed for user=%s course=%s",
+ user_id,
+ course_id,
+ )
+ return []
+
+ rows = await asyncio.to_thread(_fetch)
+ return [
+ ConceptMastery(
+ concept_name=r.get("concept_name") or "",
+ mastery=float(r.get("mastery_score") or 0.0),
+ last_reviewed_at=r.get("last_studied_at"),
+ )
+ for r in rows
+ if r.get("concept_name")
+ ]
+
+
+async def read_concepts_for_user_tool(
+ ctx: RunContext[SaplingDeps],
+) -> list[ConceptMastery]:
+ """Pydantic AI tool wrapper. Reads from ctx.deps."""
+ return await read_concepts_for_user(ctx.deps.user_id, ctx.deps.course_id)
+
+
+# ── read_misconceptions_for_course ────────────────────────────────────────
+
+
+class Misconception(BaseModel):
+ """A class-level misconception observed across student sessions."""
+
+ text: str
+ related_concept: str | None = None
+
+
+async def read_misconceptions_for_course(
+ course_id: str | None,
+) -> list[Misconception]:
+ """Return aggregated misconception strings for a course. Anonymized
+ (sourced from class-wide patterns, not from any single student).
+ Returns [] when course_id is None or the underlying table is empty.
+
+ Source: `course_concept_stats` rows for the course. Each row
+ represents one concept and carries a `common_misconceptions` array
+ (populated by the hash-gated aggregation in
+ `services/course_context_service.py`). We flatten each array entry
+ into its own Misconception, tagging `related_concept` with the
+ concept name so the agent can route distractors per-concept.
+
+ The spec referenced a hypothetical `misconceptions` table — that
+ table does not exist in this schema. `course_concept_stats` is the
+ real source of class-wide misconception strings, so we read that.
+ The tool contract (returning Misconception[]) is unchanged.
+ """
+ if not course_id:
+ return []
+
+ def _fetch() -> list[dict[str, Any]]:
+ try:
+ return (
+ table("course_concept_stats").select(
+ "concept_name,common_misconceptions",
+ filters={"course_id": f"eq.{course_id}"},
+ order="updated_at.desc",
+ limit=20,
+ )
+ or []
+ )
+ except Exception:
+ logger.exception(
+ "read_misconceptions_for_course failed for course=%s",
+ course_id,
+ )
+ return []
+
+ rows = await asyncio.to_thread(_fetch)
+ out: list[Misconception] = []
+ seen: set[str] = set()
+ for r in rows:
+ concept = r.get("concept_name") or None
+ for m in r.get("common_misconceptions") or []:
+ text = (m or "").strip() if isinstance(m, str) else ""
+ if not text:
+ continue
+ key = text.lower()
+ if key in seen:
+ continue
+ seen.add(key)
+ out.append(Misconception(text=text, related_concept=concept))
+ return out
+
+
+async def read_misconceptions_for_course_tool(
+ ctx: RunContext[SaplingDeps],
+) -> list[Misconception]:
+ """Pydantic AI tool wrapper. Reads from ctx.deps."""
+ return await read_misconceptions_for_course(ctx.deps.course_id)

backend/models/__init__.py

@@ -44,6 +44,12 @@ class GenerateQuizBody(BaseModel):
  num_questions: int = 5
  difficulty: str = "medium"
  use_shared_context: bool = True
+ # Mirrors the Learn-route fast/smart toggle so quiz generation has
+ # the same per-request model-quality choice as chat tutoring.
+ # "fast" → gemini-2.5-flash, "smart" → gemini-2.5-pro. None falls
+ # through to whatever SAPLING_MODEL_QUIZ resolves to (default
+ # gemini-2.5-flash-lite per ADR 0008).
+ model_pref: Optional[Literal["fast", "smart"]] = None
 
 
 class AnswerItem(BaseModel):

backend/agents/_providers.py

@@ -7,6 +7,7 @@
  SAPLING_MODEL_SUMMARY=gemini-2.5-flash-lite
  SAPLING_MODEL_CONCEPTS=gemini-2.5-flash
  SAPLING_MODEL_SYLLABUS=gemini-2.5-flash
+ SAPLING_MODEL_QUIZ=gemini-2.5-flash-lite
 
 Defaults are tuned per task: cheaper models for simpler classifications,
 flagship Flash for tasks where output quality drives downstream UX.
@@ -23,7 +24,7 @@
 from config import GEMINI_API_KEY
 
 
-AgentTask = Literal["classifier", "summary", "concepts", "syllabus"]
+AgentTask = Literal["classifier", "summary", "concepts", "syllabus", "quiz"]
 
 
 # Defaults are conservative. Bumping a model up costs more; the env var
@@ -33,6 +34,10 @@
  "summary": "gemini-2.5-flash-lite",
  "concepts": "gemini-2.5-flash",
  "syllabus": "gemini-2.5-flash",
+ # Quiz generation defaults to lite: it's a single-shot non-streaming
+ # call where the agent pulls structured graph data via tools, so the
+ # bulk of the value is in tool wiring, not raw model strength.
+ "quiz": "gemini-2.5-flash-lite",
 }
 
 

backend/agents/quiz.py

@@ -0,0 +1,111 @@
+"""Quiz-generation agent.
+
+Replaces routes/quiz.py:82's call_gemini_json + manual prompt-string
+augmentation. The agent has tools to pull weak concepts + class
+misconceptions on demand instead of pre-stuffing them into the prompt.
+
+Per ADR 0003 convention 4: keep the output schema compact. Gemini's
+structured-output API rejects rich nested schemas with too many states
+for serving — Quiz is a flat top-level model with a list of
+QuizQuestion items, no further nesting.
+"""
+
+from __future__ import annotations
+
+import hashlib
+from typing import Literal
+
+from pydantic import BaseModel, Field
+from pydantic_ai import Agent
+
+from agents._providers import model_for
+from agents.deps import SaplingDeps
+from agents.tools.graph_read import (
+ read_concepts_for_user_tool,
+ read_misconceptions_for_course_tool,
+)
+
+
+# Difficulty + question type are Literals so Gemini's enum constraint
+# applies and downstream UI can branch on stable strings.
+#
+# `QuizQuestionType` is intentionally MCQ-only today. The frontend
+# `submitQuiz` flow grades by `q["options"][i].correct` lookup — there's
+# no UI for free-text answers, no fuzzy-match grading, no LLM-judged
+# scoring. Generating short-answer questions through this path would
+# emit unrenderable, ungradable items. Keep the type narrow until real
+# short-answer support exists; revisit when that lands.
+QuizDifficulty = Literal["easy", "medium", "hard"]
+QuizQuestionType = Literal["multiple_choice"]
+
+
+class QuizQuestion(BaseModel):
+ """A single multiple-choice quiz question. Kept small so the parent
+ Quiz schema doesn't trip Gemini's structured-output complexity limit."""
+
+ question: str = Field(max_length=600)
+ type: QuizQuestionType
+ difficulty: QuizDifficulty
+ # 3-6 options. The agent is required to populate this for every
+ # question and to make `correct_answer` match exactly one of them.
+ options: list[str] = Field(min_length=3, max_length=6)
+ # The option text the agent considers correct. Must appear verbatim
+ # in `options`; the route validates this and drops questions that
+ # violate the contract rather than silently mis-marking them.
+ correct_answer: str = Field(max_length=400)
+ explanation: str = Field(max_length=600)
+ # Concept the question is testing — must be one of the user's known
+ # concept_names per the prompt. Used by the route to award mastery
+ # on a correct answer.
+ concept: str = Field(max_length=120)
+
+
+class Quiz(BaseModel):
+ """The agent's structured output."""
+
+ questions: list[QuizQuestion] = Field(min_length=1, max_length=20)
+
+
+_SYSTEM_PROMPT = (
+ "You generate adaptive multiple-choice quizzes for a student. Each "
+ "question must target a specific concept the student has weak "
+ "mastery on, OR address a class-level misconception you've seen.\n\n"
+ "Workflow:\n"
+ "1. Call `read_concepts_for_user` to see the student's mastery per "
+ " concept for this course (returned sorted by mastery ASC — "
+ " weakest first).\n"
+ "2. Call `read_misconceptions_for_course` to see anonymized class "
+ " misconceptions. Use these to phrase distractors and to write "
+ " a question that probes the misconception.\n"
+ "3. Compose `Quiz.questions` so the WEAKEST concepts get the most "
+ " questions, AND each item's `concept` field exactly matches a "
+ " concept_name returned by tool 1.\n\n"
+ "Per-question rules (multiple-choice only — the type field is "
+ "constrained to 'multiple_choice'):\n"
+ "- 4 options, exactly one correct. The text in `correct_answer` "
+ " MUST appear verbatim in `options` — character-for-character. "
+ " Questions that violate this are dropped at the route layer.\n"
+ "- Distractors should reflect plausible misconceptions, not random "
+ " noise. Use the read_misconceptions_for_course return value.\n"
+ "- explanation: 1-3 sentences explaining WHY the correct answer "
+ " is correct — used in the post-quiz review screen.\n"
+ "- difficulty: align with the student's mastery on the concept; "
+ " weakest concepts get easy/medium, strongest get hard.\n\n"
+ "Honor the requested num_questions and difficulty distribution "
+ "in the user message. Don't invent concepts the student doesn't "
+ "have."
+)
+_PROMPT_HASH = hashlib.sha256(_SYSTEM_PROMPT.encode("utf-8")).hexdigest()[:12]
+
+
+quiz_agent = Agent[SaplingDeps, Quiz](
+ model=model_for("quiz"),
+ deps_type=SaplingDeps,
+ output_type=Quiz,
+ system_prompt=_SYSTEM_PROMPT,
+ metadata={"prompt_version": _PROMPT_HASH, "agent": "quiz"},
+ tools=[
+ read_concepts_for_user_tool,
+ read_misconceptions_for_course_tool,
+ ],
+)

backend/agents/tools/graph_read.py

@@ -0,0 +1,163 @@
+"""Read-side graph tools for Pydantic AI agents.
+
+Per ADR 0004: the agent needs to pull mastery state + course-level
+misconceptions when planning a quiz. These are the tool surfaces.
+The pure-async functions are callable directly from routes;
+the *_tool wrappers register on a Pydantic AI Agent.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+from typing import Any
+
+from pydantic import BaseModel, Field
+from pydantic_ai import RunContext
+
+from agents.deps import SaplingDeps
+from db.connection import table
+
+logger = logging.getLogger(__name__)
+
+
+# ── read_concepts_for_user ────────────────────────────────────────────────
+
+
+class ConceptMastery(BaseModel):
+ """Per-concept mastery state for the user in a course."""
+
+ concept_name: str
+ mastery: float = Field(ge=0.0, le=1.0)
+ last_reviewed_at: str | None = None
+
+
+async def read_concepts_for_user(
+ user_id: str,
+ course_id: str | None,
+) -> list[ConceptMastery]:
+ """Return the user's concept mastery for a course (or globally if
+ course_id is None). Sorted by mastery ASC so the weakest concepts
+ appear first — quiz_agent uses this ordering to focus on weak areas.
+
+ Pure async, callable from routes. Wraps the underlying sync
+ Supabase read in asyncio.to_thread so it doesn't block the loop.
+
+ NOTE: The underlying `graph_nodes` table stores the mastery value as
+ `mastery_score` and the timestamp as `last_studied_at`. We map them
+ here to the agent-facing names (`mastery`, `last_reviewed_at`) so the
+ tool contract stays stable even if storage column names change.
+ """
+
+ def _fetch() -> list[dict[str, Any]]:
+ filters = {"user_id": f"eq.{user_id}"}
+ if course_id:
+ filters["course_id"] = f"eq.{course_id}"
+ try:
+ return (
+ table("graph_nodes").select(
+ "concept_name,mastery_score,last_studied_at",
+ filters=filters,
+ order="mastery_score.asc",
+ )
+ or []
+ )
+ except Exception:
+ logger.exception(
+ "read_concepts_for_user failed for user=%s course=%s",
+ user_id,
+ course_id,
+ )
+ return []
+
+ rows = await asyncio.to_thread(_fetch)
+ return [
+ ConceptMastery(
+ concept_name=r.get("concept_name") or "",
+ mastery=float(r.get("mastery_score") or 0.0),
+ last_reviewed_at=r.get("last_studied_at"),
+ )
+ for r in rows
+ if r.get("concept_name")
+ ]
+
+
+async def read_concepts_for_user_tool(
+ ctx: RunContext[SaplingDeps],
+) -> list[ConceptMastery]:
+ """Pydantic AI tool wrapper. Reads from ctx.deps."""
+ return await read_concepts_for_user(ctx.deps.user_id, ctx.deps.course_id)
+
+
+# ── read_misconceptions_for_course ────────────────────────────────────────
+
+
+class Misconception(BaseModel):
+ """A class-level misconception observed across student sessions."""
+
+ text: str
+ related_concept: str | None = None
+
+
+async def read_misconceptions_for_course(
+ course_id: str | None,
+) -> list[Misconception]:
+ """Return aggregated misconception strings for a course. Anonymized
+ (sourced from class-wide patterns, not from any single student).
+ Returns [] when course_id is None or the underlying table is empty.
+
+ Source: `course_concept_stats` rows for the course. Each row
+ represents one concept and carries a `common_misconceptions` array
+ (populated by the hash-gated aggregation in
+ `services/course_context_service.py`). We flatten each array entry
+ into its own Misconception, tagging `related_concept` with the
+ concept name so the agent can route distractors per-concept.
+
+ The spec referenced a hypothetical `misconceptions` table — that
+ table does not exist in this schema. `course_concept_stats` is the
+ real source of class-wide misconception strings, so we read that.
+ The tool contract (returning Misconception[]) is unchanged.
+ """
+ if not course_id:
+ return []
+
+ def _fetch() -> list[dict[str, Any]]:
+ try:
+ return (
+ table("course_concept_stats").select(
+ "concept_name,common_misconceptions",
+ filters={"course_id": f"eq.{course_id}"},
+ order="updated_at.desc",
+ limit=20,
+ )
+ or []
+ )
+ except Exception:
+ logger.exception(
+ "read_misconceptions_for_course failed for course=%s",
+ course_id,
+ )
+ return []
+
+ rows = await asyncio.to_thread(_fetch)
+ out: list[Misconception] = []
+ seen: set[str] = set()
+ for r in rows:
+ concept = r.get("concept_name") or None
+ for m in r.get("common_misconceptions") or []:
+ text = (m or "").strip() if isinstance(m, str) else ""
+ if not text:
+ continue
+ key = text.lower()
+ if key in seen:
+ continue
+ seen.add(key)
+ out.append(Misconception(text=text, related_concept=concept))
+ return out
+
+
+async def read_misconceptions_for_course_tool(
+ ctx: RunContext[SaplingDeps],
+) -> list[Misconception]:
+ """Pydantic AI tool wrapper. Reads from ctx.deps."""
+ return await read_misconceptions_for_course(ctx.deps.course_id)

backend/models/__init__.py

@@ -44,6 +44,12 @@ class GenerateQuizBody(BaseModel):
  num_questions: int = 5
  difficulty: str = "medium"
  use_shared_context: bool = True
+ # Mirrors the Learn-route fast/smart toggle so quiz generation has
+ # the same per-request model-quality choice as chat tutoring.
+ # "fast" → gemini-2.5-flash, "smart" → gemini-2.5-pro. None falls
+ # through to whatever SAPLING_MODEL_QUIZ resolves to (default
+ # gemini-2.5-flash-lite per ADR 0008).
+ model_pref: Optional[Literal["fast", "smart"]] = None
 
 
 class AnswerItem(BaseModel):

backend/agents/_providers.py

@@ -7,6 +7,7 @@
  SAPLING_MODEL_SUMMARY=gemini-2.5-flash-lite
  SAPLING_MODEL_CONCEPTS=gemini-2.5-flash
  SAPLING_MODEL_SYLLABUS=gemini-2.5-flash
+ SAPLING_MODEL_QUIZ=gemini-2.5-flash-lite
 
 Defaults are tuned per task: cheaper models for simpler classifications,
 flagship Flash for tasks where output quality drives downstream UX.
@@ -23,7 +24,7 @@
 from config import GEMINI_API_KEY
 
 
-AgentTask = Literal["classifier", "summary", "concepts", "syllabus"]
+AgentTask = Literal["classifier", "summary", "concepts", "syllabus", "quiz"]
 
 
 # Defaults are conservative. Bumping a model up costs more; the env var
@@ -33,6 +34,10 @@
  "summary": "gemini-2.5-flash-lite",
  "concepts": "gemini-2.5-flash",
  "syllabus": "gemini-2.5-flash",
+ # Quiz generation defaults to lite: it's a single-shot non-streaming
+ # call where the agent pulls structured graph data via tools, so the
+ # bulk of the value is in tool wiring, not raw model strength.
+ "quiz": "gemini-2.5-flash-lite",
 }
 
 

backend/agents/quiz.py

@@ -0,0 +1,111 @@
+"""Quiz-generation agent.
+
+Replaces routes/quiz.py:82's call_gemini_json + manual prompt-string
+augmentation. The agent has tools to pull weak concepts + class
+misconceptions on demand instead of pre-stuffing them into the prompt.
+
+Per ADR 0003 convention 4: keep the output schema compact. Gemini's
+structured-output API rejects rich nested schemas with too many states
+for serving — Quiz is a flat top-level model with a list of
+QuizQuestion items, no further nesting.
+"""
+
+from __future__ import annotations
+
+import hashlib
+from typing import Literal
+
+from pydantic import BaseModel, Field
+from pydantic_ai import Agent
+
+from agents._providers import model_for
+from agents.deps import SaplingDeps
+from agents.tools.graph_read import (
+ read_concepts_for_user_tool,
+ read_misconceptions_for_course_tool,
+)
+
+
+# Difficulty + question type are Literals so Gemini's enum constraint
+# applies and downstream UI can branch on stable strings.
+#
+# `QuizQuestionType` is intentionally MCQ-only today. The frontend
+# `submitQuiz` flow grades by `q["options"][i].correct` lookup — there's
+# no UI for free-text answers, no fuzzy-match grading, no LLM-judged
+# scoring. Generating short-answer questions through this path would
+# emit unrenderable, ungradable items. Keep the type narrow until real
+# short-answer support exists; revisit when that lands.
+QuizDifficulty = Literal["easy", "medium", "hard"]
+QuizQuestionType = Literal["multiple_choice"]
+
+
+class QuizQuestion(BaseModel):
+ """A single multiple-choice quiz question. Kept small so the parent
+ Quiz schema doesn't trip Gemini's structured-output complexity limit."""
+
+ question: str = Field(max_length=600)
+ type: QuizQuestionType
+ difficulty: QuizDifficulty
+ # 3-6 options. The agent is required to populate this for every
+ # question and to make `correct_answer` match exactly one of them.
+ options: list[str] = Field(min_length=3, max_length=6)
+ # The option text the agent considers correct. Must appear verbatim
+ # in `options`; the route validates this and drops questions that
+ # violate the contract rather than silently mis-marking them.
+ correct_answer: str = Field(max_length=400)
+ explanation: str = Field(max_length=600)
+ # Concept the question is testing — must be one of the user's known
+ # concept_names per the prompt. Used by the route to award mastery
+ # on a correct answer.
+ concept: str = Field(max_length=120)
+
+
+class Quiz(BaseModel):
+ """The agent's structured output."""
+
+ questions: list[QuizQuestion] = Field(min_length=1, max_length=20)
+
+
+_SYSTEM_PROMPT = (
+ "You generate adaptive multiple-choice quizzes for a student. Each "
+ "question must target a specific concept the student has weak "
+ "mastery on, OR address a class-level misconception you've seen.\n\n"
+ "Workflow:\n"
+ "1. Call `read_concepts_for_user` to see the student's mastery per "
+ " concept for this course (returned sorted by mastery ASC — "
+ " weakest first).\n"
+ "2. Call `read_misconceptions_for_course` to see anonymized class "
+ " misconceptions. Use these to phrase distractors and to write "
+ " a question that probes the misconception.\n"
+ "3. Compose `Quiz.questions` so the WEAKEST concepts get the most "
+ " questions, AND each item's `concept` field exactly matches a "
+ " concept_name returned by tool 1.\n\n"
+ "Per-question rules (multiple-choice only — the type field is "
+ "constrained to 'multiple_choice'):\n"
+ "- 4 options, exactly one correct. The text in `correct_answer` "
+ " MUST appear verbatim in `options` — character-for-character. "
+ " Questions that violate this are dropped at the route layer.\n"
+ "- Distractors should reflect plausible misconceptions, not random "
+ " noise. Use the read_misconceptions_for_course return value.\n"
+ "- explanation: 1-3 sentences explaining WHY the correct answer "
+ " is correct — used in the post-quiz review screen.\n"
+ "- difficulty: align with the student's mastery on the concept; "
+ " weakest concepts get easy/medium, strongest get hard.\n\n"
+ "Honor the requested num_questions and difficulty distribution "
+ "in the user message. Don't invent concepts the student doesn't "
+ "have."
+)
+_PROMPT_HASH = hashlib.sha256(_SYSTEM_PROMPT.encode("utf-8")).hexdigest()[:12]
+
+
+quiz_agent = Agent[SaplingDeps, Quiz](
+ model=model_for("quiz"),
+ deps_type=SaplingDeps,
+ output_type=Quiz,
+ system_prompt=_SYSTEM_PROMPT,
+ metadata={"prompt_version": _PROMPT_HASH, "agent": "quiz"},
+ tools=[
+ read_concepts_for_user_tool,
+ read_misconceptions_for_course_tool,
+ ],
+)

backend/agents/tools/graph_read.py

@@ -0,0 +1,163 @@
+"""Read-side graph tools for Pydantic AI agents.
+
+Per ADR 0004: the agent needs to pull mastery state + course-level
+misconceptions when planning a quiz. These are the tool surfaces.
+The pure-async functions are callable directly from routes;
+the *_tool wrappers register on a Pydantic AI Agent.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+from typing import Any
+
+from pydantic import BaseModel, Field
+from pydantic_ai import RunContext
+
+from agents.deps import SaplingDeps
+from db.connection import table
+
+logger = logging.getLogger(__name__)
+
+
+# ── read_concepts_for_user ────────────────────────────────────────────────
+
+
+class ConceptMastery(BaseModel):
+ """Per-concept mastery state for the user in a course."""
+
+ concept_name: str
+ mastery: float = Field(ge=0.0, le=1.0)
+ last_reviewed_at: str | None = None
+
+
+async def read_concepts_for_user(
+ user_id: str,
+ course_id: str | None,
+) -> list[ConceptMastery]:
+ """Return the user's concept mastery for a course (or globally if
+ course_id is None). Sorted by mastery ASC so the weakest concepts
+ appear first — quiz_agent uses this ordering to focus on weak areas.
+
+ Pure async, callable from routes. Wraps the underlying sync
+ Supabase read in asyncio.to_thread so it doesn't block the loop.
+
+ NOTE: The underlying `graph_nodes` table stores the mastery value as
+ `mastery_score` and the timestamp as `last_studied_at`. We map them
+ here to the agent-facing names (`mastery`, `last_reviewed_at`) so the
+ tool contract stays stable even if storage column names change.
+ """
+
+ def _fetch() -> list[dict[str, Any]]:
+ filters = {"user_id": f"eq.{user_id}"}
+ if course_id:
+ filters["course_id"] = f"eq.{course_id}"
+ try:
+ return (
+ table("graph_nodes").select(
+ "concept_name,mastery_score,last_studied_at",
+ filters=filters,
+ order="mastery_score.asc",
+ )
+ or []
+ )
+ except Exception:
+ logger.exception(
+ "read_concepts_for_user failed for user=%s course=%s",
+ user_id,
+ course_id,
+ )
+ return []
+
+ rows = await asyncio.to_thread(_fetch)
+ return [
+ ConceptMastery(
+ concept_name=r.get("concept_name") or "",
+ mastery=float(r.get("mastery_score") or 0.0),
+ last_reviewed_at=r.get("last_studied_at"),
+ )
+ for r in rows
+ if r.get("concept_name")
+ ]
+
+
+async def read_concepts_for_user_tool(
+ ctx: RunContext[SaplingDeps],
+) -> list[ConceptMastery]:
+ """Pydantic AI tool wrapper. Reads from ctx.deps."""
+ return await read_concepts_for_user(ctx.deps.user_id, ctx.deps.course_id)
+
+
+# ── read_misconceptions_for_course ────────────────────────────────────────
+
+
+class Misconception(BaseModel):
+ """A class-level misconception observed across student sessions."""
+
+ text: str
+ related_concept: str | None = None
+
+
+async def read_misconceptions_for_course(
+ course_id: str | None,
+) -> list[Misconception]:
+ """Return aggregated misconception strings for a course. Anonymized
+ (sourced from class-wide patterns, not from any single student).
+ Returns [] when course_id is None or the underlying table is empty.
+
+ Source: `course_concept_stats` rows for the course. Each row
+ represents one concept and carries a `common_misconceptions` array
+ (populated by the hash-gated aggregation in
+ `services/course_context_service.py`). We flatten each array entry
+ into its own Misconception, tagging `related_concept` with the
+ concept name so the agent can route distractors per-concept.
+
+ The spec referenced a hypothetical `misconceptions` table — that
+ table does not exist in this schema. `course_concept_stats` is the
+ real source of class-wide misconception strings, so we read that.
+ The tool contract (returning Misconception[]) is unchanged.
+ """
+ if not course_id:
+ return []
+
+ def _fetch() -> list[dict[str, Any]]:
+ try:
+ return (
+ table("course_concept_stats").select(
+ "concept_name,common_misconceptions",
+ filters={"course_id": f"eq.{course_id}"},
+ order="updated_at.desc",
+ limit=20,
+ )
+ or []
+ )
+ except Exception:
+ logger.exception(
+ "read_misconceptions_for_course failed for course=%s",
+ course_id,
+ )
+ return []
+
+ rows = await asyncio.to_thread(_fetch)
+ out: list[Misconception] = []
+ seen: set[str] = set()
+ for r in rows:
+ concept = r.get("concept_name") or None
+ for m in r.get("common_misconceptions") or []:
+ text = (m or "").strip() if isinstance(m, str) else ""
+ if not text:
+ continue
+ key = text.lower()
+ if key in seen:
+ continue
+ seen.add(key)
+ out.append(Misconception(text=text, related_concept=concept))
+ return out
+
+
+async def read_misconceptions_for_course_tool(
+ ctx: RunContext[SaplingDeps],
+) -> list[Misconception]:
+ """Pydantic AI tool wrapper. Reads from ctx.deps."""
+ return await read_misconceptions_for_course(ctx.deps.course_id)

backend/models/__init__.py

@@ -44,6 +44,12 @@ class GenerateQuizBody(BaseModel):
  num_questions: int = 5
  difficulty: str = "medium"
  use_shared_context: bool = True
+ # Mirrors the Learn-route fast/smart toggle so quiz generation has
+ # the same per-request model-quality choice as chat tutoring.
+ # "fast" → gemini-2.5-flash, "smart" → gemini-2.5-pro. None falls
+ # through to whatever SAPLING_MODEL_QUIZ resolves to (default
+ # gemini-2.5-flash-lite per ADR 0008).
+ model_pref: Optional[Literal["fast", "smart"]] = None
 
 
 class AnswerItem(BaseModel):

backend/agents/_providers.py

@@ -7,6 +7,7 @@
  SAPLING_MODEL_SUMMARY=gemini-2.5-flash-lite
  SAPLING_MODEL_CONCEPTS=gemini-2.5-flash
  SAPLING_MODEL_SYLLABUS=gemini-2.5-flash
+ SAPLING_MODEL_QUIZ=gemini-2.5-flash-lite
 
 Defaults are tuned per task: cheaper models for simpler classifications,
 flagship Flash for tasks where output quality drives downstream UX.
@@ -23,7 +24,7 @@
 from config import GEMINI_API_KEY
 
 
-AgentTask = Literal["classifier", "summary", "concepts", "syllabus"]
+AgentTask = Literal["classifier", "summary", "concepts", "syllabus", "quiz"]
 
 
 # Defaults are conservative. Bumping a model up costs more; the env var
@@ -33,6 +34,10 @@
  "summary": "gemini-2.5-flash-lite",
  "concepts": "gemini-2.5-flash",
  "syllabus": "gemini-2.5-flash",
+ # Quiz generation defaults to lite: it's a single-shot non-streaming
+ # call where the agent pulls structured graph data via tools, so the
+ # bulk of the value is in tool wiring, not raw model strength.
+ "quiz": "gemini-2.5-flash-lite",
 }
 
 

backend/agents/quiz.py

@@ -0,0 +1,111 @@
+"""Quiz-generation agent.
+
+Replaces routes/quiz.py:82's call_gemini_json + manual prompt-string
+augmentation. The agent has tools to pull weak concepts + class
+misconceptions on demand instead of pre-stuffing them into the prompt.
+
+Per ADR 0003 convention 4: keep the output schema compact. Gemini's
+structured-output API rejects rich nested schemas with too many states
+for serving — Quiz is a flat top-level model with a list of
+QuizQuestion items, no further nesting.
+"""
+
+from __future__ import annotations
+
+import hashlib
+from typing import Literal
+
+from pydantic import BaseModel, Field
+from pydantic_ai import Agent
+
+from agents._providers import model_for
+from agents.deps import SaplingDeps
+from agents.tools.graph_read import (
+ read_concepts_for_user_tool,
+ read_misconceptions_for_course_tool,
+)
+
+
+# Difficulty + question type are Literals so Gemini's enum constraint
+# applies and downstream UI can branch on stable strings.
+#
+# `QuizQuestionType` is intentionally MCQ-only today. The frontend
+# `submitQuiz` flow grades by `q["options"][i].correct` lookup — there's
+# no UI for free-text answers, no fuzzy-match grading, no LLM-judged
+# scoring. Generating short-answer questions through this path would
+# emit unrenderable, ungradable items. Keep the type narrow until real
+# short-answer support exists; revisit when that lands.
+QuizDifficulty = Literal["easy", "medium", "hard"]
+QuizQuestionType = Literal["multiple_choice"]
+
+
+class QuizQuestion(BaseModel):
+ """A single multiple-choice quiz question. Kept small so the parent
+ Quiz schema doesn't trip Gemini's structured-output complexity limit."""
+
+ question: str = Field(max_length=600)
+ type: QuizQuestionType
+ difficulty: QuizDifficulty
+ # 3-6 options. The agent is required to populate this for every
+ # question and to make `correct_answer` match exactly one of them.
+ options: list[str] = Field(min_length=3, max_length=6)
+ # The option text the agent considers correct. Must appear verbatim
+ # in `options`; the route validates this and drops questions that
+ # violate the contract rather than silently mis-marking them.
+ correct_answer: str = Field(max_length=400)
+ explanation: str = Field(max_length=600)
+ # Concept the question is testing — must be one of the user's known
+ # concept_names per the prompt. Used by the route to award mastery
+ # on a correct answer.
+ concept: str = Field(max_length=120)
+
+
+class Quiz(BaseModel):
+ """The agent's structured output."""
+
+ questions: list[QuizQuestion] = Field(min_length=1, max_length=20)
+
+
+_SYSTEM_PROMPT = (
+ "You generate adaptive multiple-choice quizzes for a student. Each "
+ "question must target a specific concept the student has weak "
+ "mastery on, OR address a class-level misconception you've seen.\n\n"
+ "Workflow:\n"
+ "1. Call `read_concepts_for_user` to see the student's mastery per "
+ " concept for this course (returned sorted by mastery ASC — "
+ " weakest first).\n"
+ "2. Call `read_misconceptions_for_course` to see anonymized class "
+ " misconceptions. Use these to phrase distractors and to write "
+ " a question that probes the misconception.\n"
+ "3. Compose `Quiz.questions` so the WEAKEST concepts get the most "
+ " questions, AND each item's `concept` field exactly matches a "
+ " concept_name returned by tool 1.\n\n"
+ "Per-question rules (multiple-choice only — the type field is "
+ "constrained to 'multiple_choice'):\n"
+ "- 4 options, exactly one correct. The text in `correct_answer` "
+ " MUST appear verbatim in `options` — character-for-character. "
+ " Questions that violate this are dropped at the route layer.\n"
+ "- Distractors should reflect plausible misconceptions, not random "
+ " noise. Use the read_misconceptions_for_course return value.\n"
+ "- explanation: 1-3 sentences explaining WHY the correct answer "
+ " is correct — used in the post-quiz review screen.\n"
+ "- difficulty: align with the student's mastery on the concept; "
+ " weakest concepts get easy/medium, strongest get hard.\n\n"
+ "Honor the requested num_questions and difficulty distribution "
+ "in the user message. Don't invent concepts the student doesn't "
+ "have."
+)
+_PROMPT_HASH = hashlib.sha256(_SYSTEM_PROMPT.encode("utf-8")).hexdigest()[:12]
+
+
+quiz_agent = Agent[SaplingDeps, Quiz](
+ model=model_for("quiz"),
+ deps_type=SaplingDeps,
+ output_type=Quiz,
+ system_prompt=_SYSTEM_PROMPT,
+ metadata={"prompt_version": _PROMPT_HASH, "agent": "quiz"},
+ tools=[
+ read_concepts_for_user_tool,
+ read_misconceptions_for_course_tool,
+ ],
+)

backend/agents/tools/graph_read.py

@@ -0,0 +1,163 @@
+"""Read-side graph tools for Pydantic AI agents.
+
+Per ADR 0004: the agent needs to pull mastery state + course-level
+misconceptions when planning a quiz. These are the tool surfaces.
+The pure-async functions are callable directly from routes;
+the *_tool wrappers register on a Pydantic AI Agent.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+from typing import Any
+
+from pydantic import BaseModel, Field
+from pydantic_ai import RunContext
+
+from agents.deps import SaplingDeps
+from db.connection import table
+
+logger = logging.getLogger(__name__)
+
+
+# ── read_concepts_for_user ────────────────────────────────────────────────
+
+
+class ConceptMastery(BaseModel):
+ """Per-concept mastery state for the user in a course."""
+
+ concept_name: str
+ mastery: float = Field(ge=0.0, le=1.0)
+ last_reviewed_at: str | None = None
+
+
+async def read_concepts_for_user(
+ user_id: str,
+ course_id: str | None,
+) -> list[ConceptMastery]:
+ """Return the user's concept mastery for a course (or globally if
+ course_id is None). Sorted by mastery ASC so the weakest concepts
+ appear first — quiz_agent uses this ordering to focus on weak areas.
+
+ Pure async, callable from routes. Wraps the underlying sync
+ Supabase read in asyncio.to_thread so it doesn't block the loop.
+
+ NOTE: The underlying `graph_nodes` table stores the mastery value as
+ `mastery_score` and the timestamp as `last_studied_at`. We map them
+ here to the agent-facing names (`mastery`, `last_reviewed_at`) so the
+ tool contract stays stable even if storage column names change.
+ """
+
+ def _fetch() -> list[dict[str, Any]]:
+ filters = {"user_id": f"eq.{user_id}"}
+ if course_id:
+ filters["course_id"] = f"eq.{course_id}"
+ try:
+ return (
+ table("graph_nodes").select(
+ "concept_name,mastery_score,last_studied_at",
+ filters=filters,
+ order="mastery_score.asc",
+ )
+ or []
+ )
+ except Exception:
+ logger.exception(
+ "read_concepts_for_user failed for user=%s course=%s",
+ user_id,
+ course_id,
+ )
+ return []
+
+ rows = await asyncio.to_thread(_fetch)
+ return [
+ ConceptMastery(
+ concept_name=r.get("concept_name") or "",
+ mastery=float(r.get("mastery_score") or 0.0),
+ last_reviewed_at=r.get("last_studied_at"),
+ )
+ for r in rows
+ if r.get("concept_name")
+ ]
+
+
+async def read_concepts_for_user_tool(
+ ctx: RunContext[SaplingDeps],
+) -> list[ConceptMastery]:
+ """Pydantic AI tool wrapper. Reads from ctx.deps."""
+ return await read_concepts_for_user(ctx.deps.user_id, ctx.deps.course_id)
+
+
+# ── read_misconceptions_for_course ────────────────────────────────────────
+
+
+class Misconception(BaseModel):
+ """A class-level misconception observed across student sessions."""
+
+ text: str
+ related_concept: str | None = None
+
+
+async def read_misconceptions_for_course(
+ course_id: str | None,
+) -> list[Misconception]:
+ """Return aggregated misconception strings for a course. Anonymized
+ (sourced from class-wide patterns, not from any single student).
+ Returns [] when course_id is None or the underlying table is empty.
+
+ Source: `course_concept_stats` rows for the course. Each row
+ represents one concept and carries a `common_misconceptions` array
+ (populated by the hash-gated aggregation in
+ `services/course_context_service.py`). We flatten each array entry
+ into its own Misconception, tagging `related_concept` with the
+ concept name so the agent can route distractors per-concept.
+
+ The spec referenced a hypothetical `misconceptions` table — that
+ table does not exist in this schema. `course_concept_stats` is the
+ real source of class-wide misconception strings, so we read that.
+ The tool contract (returning Misconception[]) is unchanged.
+ """
+ if not course_id:
+ return []
+
+ def _fetch() -> list[dict[str, Any]]:
+ try:
+ return (
+ table("course_concept_stats").select(
+ "concept_name,common_misconceptions",
+ filters={"course_id": f"eq.{course_id}"},
+ order="updated_at.desc",
+ limit=20,
+ )
+ or []
+ )
+ except Exception:
+ logger.exception(
+ "read_misconceptions_for_course failed for course=%s",
+ course_id,
+ )
+ return []
+
+ rows = await asyncio.to_thread(_fetch)
+ out: list[Misconception] = []
+ seen: set[str] = set()
+ for r in rows:
+ concept = r.get("concept_name") or None
+ for m in r.get("common_misconceptions") or []:
+ text = (m or "").strip() if isinstance(m, str) else ""
+ if not text:
+ continue
+ key = text.lower()
+ if key in seen:
+ continue
+ seen.add(key)
+ out.append(Misconception(text=text, related_concept=concept))
+ return out
+
+
+async def read_misconceptions_for_course_tool(
+ ctx: RunContext[SaplingDeps],
+) -> list[Misconception]:
+ """Pydantic AI tool wrapper. Reads from ctx.deps."""
+ return await read_misconceptions_for_course(ctx.deps.course_id)

backend/models/__init__.py

@@ -44,6 +44,12 @@ class GenerateQuizBody(BaseModel):
  num_questions: int = 5
  difficulty: str = "medium"
  use_shared_context: bool = True
+ # Mirrors the Learn-route fast/smart toggle so quiz generation has
+ # the same per-request model-quality choice as chat tutoring.
+ # "fast" → gemini-2.5-flash, "smart" → gemini-2.5-pro. None falls
+ # through to whatever SAPLING_MODEL_QUIZ resolves to (default
+ # gemini-2.5-flash-lite per ADR 0008).
+ model_pref: Optional[Literal["fast", "smart"]] = None
 
 
 class AnswerItem(BaseModel):

backend/agents/_providers.py

@@ -7,6 +7,7 @@
  SAPLING_MODEL_SUMMARY=gemini-2.5-flash-lite
  SAPLING_MODEL_CONCEPTS=gemini-2.5-flash
  SAPLING_MODEL_SYLLABUS=gemini-2.5-flash
+ SAPLING_MODEL_QUIZ=gemini-2.5-flash-lite
 
 Defaults are tuned per task: cheaper models for simpler classifications,
 flagship Flash for tasks where output quality drives downstream UX.
@@ -23,7 +24,7 @@
 from config import GEMINI_API_KEY
 
 
-AgentTask = Literal["classifier", "summary", "concepts", "syllabus"]
+AgentTask = Literal["classifier", "summary", "concepts", "syllabus", "quiz"]
 
 
 # Defaults are conservative. Bumping a model up costs more; the env var
@@ -33,6 +34,10 @@
  "summary": "gemini-2.5-flash-lite",
  "concepts": "gemini-2.5-flash",
  "syllabus": "gemini-2.5-flash",
+ # Quiz generation defaults to lite: it's a single-shot non-streaming
+ # call where the agent pulls structured graph data via tools, so the
+ # bulk of the value is in tool wiring, not raw model strength.
+ "quiz": "gemini-2.5-flash-lite",
 }
 
 

backend/agents/quiz.py

@@ -0,0 +1,111 @@
+"""Quiz-generation agent.
+
+Replaces routes/quiz.py:82's call_gemini_json + manual prompt-string
+augmentation. The agent has tools to pull weak concepts + class
+misconceptions on demand instead of pre-stuffing them into the prompt.
+
+Per ADR 0003 convention 4: keep the output schema compact. Gemini's
+structured-output API rejects rich nested schemas with too many states
+for serving — Quiz is a flat top-level model with a list of
+QuizQuestion items, no further nesting.
+"""
+
+from __future__ import annotations
+
+import hashlib
+from typing import Literal
+
+from pydantic import BaseModel, Field
+from pydantic_ai import Agent
+
+from agents._providers import model_for
+from agents.deps import SaplingDeps
+from agents.tools.graph_read import (
+ read_concepts_for_user_tool,
+ read_misconceptions_for_course_tool,
+)
+
+
+# Difficulty + question type are Literals so Gemini's enum constraint
+# applies and downstream UI can branch on stable strings.
+#
+# `QuizQuestionType` is intentionally MCQ-only today. The frontend
+# `submitQuiz` flow grades by `q["options"][i].correct` lookup — there's
+# no UI for free-text answers, no fuzzy-match grading, no LLM-judged
+# scoring. Generating short-answer questions through this path would
+# emit unrenderable, ungradable items. Keep the type narrow until real
+# short-answer support exists; revisit when that lands.
+QuizDifficulty = Literal["easy", "medium", "hard"]
+QuizQuestionType = Literal["multiple_choice"]
+
+
+class QuizQuestion(BaseModel):
+ """A single multiple-choice quiz question. Kept small so the parent
+ Quiz schema doesn't trip Gemini's structured-output complexity limit."""
+
+ question: str = Field(max_length=600)
+ type: QuizQuestionType
+ difficulty: QuizDifficulty
+ # 3-6 options. The agent is required to populate this for every
+ # question and to make `correct_answer` match exactly one of them.
+ options: list[str] = Field(min_length=3, max_length=6)
+ # The option text the agent considers correct. Must appear verbatim
+ # in `options`; the route validates this and drops questions that
+ # violate the contract rather than silently mis-marking them.
+ correct_answer: str = Field(max_length=400)
+ explanation: str = Field(max_length=600)
+ # Concept the question is testing — must be one of the user's known
+ # concept_names per the prompt. Used by the route to award mastery
+ # on a correct answer.
+ concept: str = Field(max_length=120)
+
+
+class Quiz(BaseModel):
+ """The agent's structured output."""
+
+ questions: list[QuizQuestion] = Field(min_length=1, max_length=20)
+
+
+_SYSTEM_PROMPT = (
+ "You generate adaptive multiple-choice quizzes for a student. Each "
+ "question must target a specific concept the student has weak "
+ "mastery on, OR address a class-level misconception you've seen.\n\n"
+ "Workflow:\n"
+ "1. Call `read_concepts_for_user` to see the student's mastery per "
+ " concept for this course (returned sorted by mastery ASC — "
+ " weakest first).\n"
+ "2. Call `read_misconceptions_for_course` to see anonymized class "
+ " misconceptions. Use these to phrase distractors and to write "
+ " a question that probes the misconception.\n"
+ "3. Compose `Quiz.questions` so the WEAKEST concepts get the most "
+ " questions, AND each item's `concept` field exactly matches a "
+ " concept_name returned by tool 1.\n\n"
+ "Per-question rules (multiple-choice only — the type field is "
+ "constrained to 'multiple_choice'):\n"
+ "- 4 options, exactly one correct. The text in `correct_answer` "
+ " MUST appear verbatim in `options` — character-for-character. "
+ " Questions that violate this are dropped at the route layer.\n"
+ "- Distractors should reflect plausible misconceptions, not random "
+ " noise. Use the read_misconceptions_for_course return value.\n"
+ "- explanation: 1-3 sentences explaining WHY the correct answer "
+ " is correct — used in the post-quiz review screen.\n"
+ "- difficulty: align with the student's mastery on the concept; "
+ " weakest concepts get easy/medium, strongest get hard.\n\n"
+ "Honor the requested num_questions and difficulty distribution "
+ "in the user message. Don't invent concepts the student doesn't "
+ "have."
+)
+_PROMPT_HASH = hashlib.sha256(_SYSTEM_PROMPT.encode("utf-8")).hexdigest()[:12]
+
+
+quiz_agent = Agent[SaplingDeps, Quiz](
+ model=model_for("quiz"),
+ deps_type=SaplingDeps,
+ output_type=Quiz,
+ system_prompt=_SYSTEM_PROMPT,
+ metadata={"prompt_version": _PROMPT_HASH, "agent": "quiz"},
+ tools=[
+ read_concepts_for_user_tool,
+ read_misconceptions_for_course_tool,
+ ],
+)

backend/agents/tools/graph_read.py

@@ -0,0 +1,163 @@
+"""Read-side graph tools for Pydantic AI agents.
+
+Per ADR 0004: the agent needs to pull mastery state + course-level
+misconceptions when planning a quiz. These are the tool surfaces.
+The pure-async functions are callable directly from routes;
+the *_tool wrappers register on a Pydantic AI Agent.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+from typing import Any
+
+from pydantic import BaseModel, Field
+from pydantic_ai import RunContext
+
+from agents.deps import SaplingDeps
+from db.connection import table
+
+logger = logging.getLogger(__name__)
+
+
+# ── read_concepts_for_user ────────────────────────────────────────────────
+
+
+class ConceptMastery(BaseModel):
+ """Per-concept mastery state for the user in a course."""
+
+ concept_name: str
+ mastery: float = Field(ge=0.0, le=1.0)
+ last_reviewed_at: str | None = None
+
+
+async def read_concepts_for_user(
+ user_id: str,
+ course_id: str | None,
+) -> list[ConceptMastery]:
+ """Return the user's concept mastery for a course (or globally if
+ course_id is None). Sorted by mastery ASC so the weakest concepts
+ appear first — quiz_agent uses this ordering to focus on weak areas.
+
+ Pure async, callable from routes. Wraps the underlying sync
+ Supabase read in asyncio.to_thread so it doesn't block the loop.
+
+ NOTE: The underlying `graph_nodes` table stores the mastery value as
+ `mastery_score` and the timestamp as `last_studied_at`. We map them
+ here to the agent-facing names (`mastery`, `last_reviewed_at`) so the
+ tool contract stays stable even if storage column names change.
+ """
+
+ def _fetch() -> list[dict[str, Any]]:
+ filters = {"user_id": f"eq.{user_id}"}
+ if course_id:
+ filters["course_id"] = f"eq.{course_id}"
+ try:
+ return (
+ table("graph_nodes").select(
+ "concept_name,mastery_score,last_studied_at",
+ filters=filters,
+ order="mastery_score.asc",
+ )
+ or []
+ )
+ except Exception:
+ logger.exception(
+ "read_concepts_for_user failed for user=%s course=%s",
+ user_id,
+ course_id,
+ )
+ return []
+
+ rows = await asyncio.to_thread(_fetch)
+ return [
+ ConceptMastery(
+ concept_name=r.get("concept_name") or "",
+ mastery=float(r.get("mastery_score") or 0.0),
+ last_reviewed_at=r.get("last_studied_at"),
+ )
+ for r in rows
+ if r.get("concept_name")
+ ]
+
+
+async def read_concepts_for_user_tool(
+ ctx: RunContext[SaplingDeps],
+) -> list[ConceptMastery]:
+ """Pydantic AI tool wrapper. Reads from ctx.deps."""
+ return await read_concepts_for_user(ctx.deps.user_id, ctx.deps.course_id)
+
+
+# ── read_misconceptions_for_course ────────────────────────────────────────
+
+
+class Misconception(BaseModel):
+ """A class-level misconception observed across student sessions."""
+
+ text: str
+ related_concept: str | None = None
+
+
+async def read_misconceptions_for_course(
+ course_id: str | None,
+) -> list[Misconception]:
+ """Return aggregated misconception strings for a course. Anonymized
+ (sourced from class-wide patterns, not from any single student).
+ Returns [] when course_id is None or the underlying table is empty.
+
+ Source: `course_concept_stats` rows for the course. Each row
+ represents one concept and carries a `common_misconceptions` array
+ (populated by the hash-gated aggregation in
+ `services/course_context_service.py`). We flatten each array entry
+ into its own Misconception, tagging `related_concept` with the
+ concept name so the agent can route distractors per-concept.
+
+ The spec referenced a hypothetical `misconceptions` table — that
+ table does not exist in this schema. `course_concept_stats` is the
+ real source of class-wide misconception strings, so we read that.
+ The tool contract (returning Misconception[]) is unchanged.
+ """
+ if not course_id:
+ return []
+
+ def _fetch() -> list[dict[str, Any]]:
+ try:
+ return (
+ table("course_concept_stats").select(
+ "concept_name,common_misconceptions",
+ filters={"course_id": f"eq.{course_id}"},
+ order="updated_at.desc",
+ limit=20,
+ )
+ or []
+ )
+ except Exception:
+ logger.exception(
+ "read_misconceptions_for_course failed for course=%s",
+ course_id,
+ )
+ return []
+
+ rows = await asyncio.to_thread(_fetch)
+ out: list[Misconception] = []
+ seen: set[str] = set()
+ for r in rows:
+ concept = r.get("concept_name") or None
+ for m in r.get("common_misconceptions") or []:
+ text = (m or "").strip() if isinstance(m, str) else ""
+ if not text:
+ continue
+ key = text.lower()
+ if key in seen:
+ continue
+ seen.add(key)
+ out.append(Misconception(text=text, related_concept=concept))
+ return out
+
+
+async def read_misconceptions_for_course_tool(
+ ctx: RunContext[SaplingDeps],
+) -> list[Misconception]:
+ """Pydantic AI tool wrapper. Reads from ctx.deps."""
+ return await read_misconceptions_for_course(ctx.deps.course_id)

backend/models/__init__.py

@@ -44,6 +44,12 @@ class GenerateQuizBody(BaseModel):
  num_questions: int = 5
  difficulty: str = "medium"
  use_shared_context: bool = True
+ # Mirrors the Learn-route fast/smart toggle so quiz generation has
+ # the same per-request model-quality choice as chat tutoring.
+ # "fast" → gemini-2.5-flash, "smart" → gemini-2.5-pro. None falls
+ # through to whatever SAPLING_MODEL_QUIZ resolves to (default
+ # gemini-2.5-flash-lite per ADR 0008).
+ model_pref: Optional[Literal["fast", "smart"]] = None
 
 
 class AnswerItem(BaseModel):

backend/agents/_providers.py

@@ -7,6 +7,7 @@
  SAPLING_MODEL_SUMMARY=gemini-2.5-flash-lite
  SAPLING_MODEL_CONCEPTS=gemini-2.5-flash
  SAPLING_MODEL_SYLLABUS=gemini-2.5-flash
+ SAPLING_MODEL_QUIZ=gemini-2.5-flash-lite
 
 Defaults are tuned per task: cheaper models for simpler classifications,
 flagship Flash for tasks where output quality drives downstream UX.
@@ -23,7 +24,7 @@
 from config import GEMINI_API_KEY
 
 
-AgentTask = Literal["classifier", "summary", "concepts", "syllabus"]
+AgentTask = Literal["classifier", "summary", "concepts", "syllabus", "quiz"]
 
 
 # Defaults are conservative. Bumping a model up costs more; the env var
@@ -33,6 +34,10 @@
  "summary": "gemini-2.5-flash-lite",
  "concepts": "gemini-2.5-flash",
  "syllabus": "gemini-2.5-flash",
+ # Quiz generation defaults to lite: it's a single-shot non-streaming
+ # call where the agent pulls structured graph data via tools, so the
+ # bulk of the value is in tool wiring, not raw model strength.
+ "quiz": "gemini-2.5-flash-lite",
 }
 
 

backend/agents/quiz.py

@@ -0,0 +1,111 @@
+"""Quiz-generation agent.
+
+Replaces routes/quiz.py:82's call_gemini_json + manual prompt-string
+augmentation. The agent has tools to pull weak concepts + class
+misconceptions on demand instead of pre-stuffing them into the prompt.
+
+Per ADR 0003 convention 4: keep the output schema compact. Gemini's
+structured-output API rejects rich nested schemas with too many states
+for serving — Quiz is a flat top-level model with a list of
+QuizQuestion items, no further nesting.
+"""
+
+from __future__ import annotations
+
+import hashlib
+from typing import Literal
+
+from pydantic import BaseModel, Field
+from pydantic_ai import Agent
+
+from agents._providers import model_for
+from agents.deps import SaplingDeps
+from agents.tools.graph_read import (
+ read_concepts_for_user_tool,
+ read_misconceptions_for_course_tool,
+)
+
+
+# Difficulty + question type are Literals so Gemini's enum constraint
+# applies and downstream UI can branch on stable strings.
+#
+# `QuizQuestionType` is intentionally MCQ-only today. The frontend
+# `submitQuiz` flow grades by `q["options"][i].correct` lookup — there's
+# no UI for free-text answers, no fuzzy-match grading, no LLM-judged
+# scoring. Generating short-answer questions through this path would
+# emit unrenderable, ungradable items. Keep the type narrow until real
+# short-answer support exists; revisit when that lands.
+QuizDifficulty = Literal["easy", "medium", "hard"]
+QuizQuestionType = Literal["multiple_choice"]
+
+
+class QuizQuestion(BaseModel):
+ """A single multiple-choice quiz question. Kept small so the parent
+ Quiz schema doesn't trip Gemini's structured-output complexity limit."""
+
+ question: str = Field(max_length=600)
+ type: QuizQuestionType
+ difficulty: QuizDifficulty
+ # 3-6 options. The agent is required to populate this for every
+ # question and to make `correct_answer` match exactly one of them.
+ options: list[str] = Field(min_length=3, max_length=6)
+ # The option text the agent considers correct. Must appear verbatim
+ # in `options`; the route validates this and drops questions that
+ # violate the contract rather than silently mis-marking them.
+ correct_answer: str = Field(max_length=400)
+ explanation: str = Field(max_length=600)
+ # Concept the question is testing — must be one of the user's known
+ # concept_names per the prompt. Used by the route to award mastery
+ # on a correct answer.
+ concept: str = Field(max_length=120)
+
+
+class Quiz(BaseModel):
+ """The agent's structured output."""
+
+ questions: list[QuizQuestion] = Field(min_length=1, max_length=20)
+
+
+_SYSTEM_PROMPT = (
+ "You generate adaptive multiple-choice quizzes for a student. Each "
+ "question must target a specific concept the student has weak "
+ "mastery on, OR address a class-level misconception you've seen.\n\n"
+ "Workflow:\n"
+ "1. Call `read_concepts_for_user` to see the student's mastery per "
+ " concept for this course (returned sorted by mastery ASC — "
+ " weakest first).\n"
+ "2. Call `read_misconceptions_for_course` to see anonymized class "
+ " misconceptions. Use these to phrase distractors and to write "
+ " a question that probes the misconception.\n"
+ "3. Compose `Quiz.questions` so the WEAKEST concepts get the most "
+ " questions, AND each item's `concept` field exactly matches a "
+ " concept_name returned by tool 1.\n\n"
+ "Per-question rules (multiple-choice only — the type field is "
+ "constrained to 'multiple_choice'):\n"
+ "- 4 options, exactly one correct. The text in `correct_answer` "
+ " MUST appear verbatim in `options` — character-for-character. "
+ " Questions that violate this are dropped at the route layer.\n"
+ "- Distractors should reflect plausible misconceptions, not random "
+ " noise. Use the read_misconceptions_for_course return value.\n"
+ "- explanation: 1-3 sentences explaining WHY the correct answer "
+ " is correct — used in the post-quiz review screen.\n"
+ "- difficulty: align with the student's mastery on the concept; "
+ " weakest concepts get easy/medium, strongest get hard.\n\n"
+ "Honor the requested num_questions and difficulty distribution "
+ "in the user message. Don't invent concepts the student doesn't "
+ "have."
+)
+_PROMPT_HASH = hashlib.sha256(_SYSTEM_PROMPT.encode("utf-8")).hexdigest()[:12]
+
+
+quiz_agent = Agent[SaplingDeps, Quiz](
+ model=model_for("quiz"),
+ deps_type=SaplingDeps,
+ output_type=Quiz,
+ system_prompt=_SYSTEM_PROMPT,
+ metadata={"prompt_version": _PROMPT_HASH, "agent": "quiz"},
+ tools=[
+ read_concepts_for_user_tool,
+ read_misconceptions_for_course_tool,
+ ],
+)

backend/agents/tools/graph_read.py

@@ -0,0 +1,163 @@
+"""Read-side graph tools for Pydantic AI agents.
+
+Per ADR 0004: the agent needs to pull mastery state + course-level
+misconceptions when planning a quiz. These are the tool surfaces.
+The pure-async functions are callable directly from routes;
+the *_tool wrappers register on a Pydantic AI Agent.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+from typing import Any
+
+from pydantic import BaseModel, Field
+from pydantic_ai import RunContext
+
+from agents.deps import SaplingDeps
+from db.connection import table
+
+logger = logging.getLogger(__name__)
+
+
+# ── read_concepts_for_user ────────────────────────────────────────────────
+
+
+class ConceptMastery(BaseModel):
+ """Per-concept mastery state for the user in a course."""
+
+ concept_name: str
+ mastery: float = Field(ge=0.0, le=1.0)
+ last_reviewed_at: str | None = None
+
+
+async def read_concepts_for_user(
+ user_id: str,
+ course_id: str | None,
+) -> list[ConceptMastery]:
+ """Return the user's concept mastery for a course (or globally if
+ course_id is None). Sorted by mastery ASC so the weakest concepts
+ appear first — quiz_agent uses this ordering to focus on weak areas.
+
+ Pure async, callable from routes. Wraps the underlying sync
+ Supabase read in asyncio.to_thread so it doesn't block the loop.
+
+ NOTE: The underlying `graph_nodes` table stores the mastery value as
+ `mastery_score` and the timestamp as `last_studied_at`. We map them
+ here to the agent-facing names (`mastery`, `last_reviewed_at`) so the
+ tool contract stays stable even if storage column names change.
+ """
+
+ def _fetch() -> list[dict[str, Any]]:
+ filters = {"user_id": f"eq.{user_id}"}
+ if course_id:
+ filters["course_id"] = f"eq.{course_id}"
+ try:
+ return (
+ table("graph_nodes").select(
+ "concept_name,mastery_score,last_studied_at",
+ filters=filters,
+ order="mastery_score.asc",
+ )
+ or []
+ )
+ except Exception:
+ logger.exception(
+ "read_concepts_for_user failed for user=%s course=%s",
+ user_id,
+ course_id,
+ )
+ return []
+
+ rows = await asyncio.to_thread(_fetch)
+ return [
+ ConceptMastery(
+ concept_name=r.get("concept_name") or "",
+ mastery=float(r.get("mastery_score") or 0.0),
+ last_reviewed_at=r.get("last_studied_at"),
+ )
+ for r in rows
+ if r.get("concept_name")
+ ]
+
+
+async def read_concepts_for_user_tool(
+ ctx: RunContext[SaplingDeps],
+) -> list[ConceptMastery]:
+ """Pydantic AI tool wrapper. Reads from ctx.deps."""
+ return await read_concepts_for_user(ctx.deps.user_id, ctx.deps.course_id)
+
+
+# ── read_misconceptions_for_course ────────────────────────────────────────
+
+
+class Misconception(BaseModel):
+ """A class-level misconception observed across student sessions."""
+
+ text: str
+ related_concept: str | None = None
+
+
+async def read_misconceptions_for_course(
+ course_id: str | None,
+) -> list[Misconception]:
+ """Return aggregated misconception strings for a course. Anonymized
+ (sourced from class-wide patterns, not from any single student).
+ Returns [] when course_id is None or the underlying table is empty.
+
+ Source: `course_concept_stats` rows for the course. Each row
+ represents one concept and carries a `common_misconceptions` array
+ (populated by the hash-gated aggregation in
+ `services/course_context_service.py`). We flatten each array entry
+ into its own Misconception, tagging `related_concept` with the
+ concept name so the agent can route distractors per-concept.
+
+ The spec referenced a hypothetical `misconceptions` table — that
+ table does not exist in this schema. `course_concept_stats` is the
+ real source of class-wide misconception strings, so we read that.
+ The tool contract (returning Misconception[]) is unchanged.
+ """
+ if not course_id:
+ return []
+
+ def _fetch() -> list[dict[str, Any]]:
+ try:
+ return (
+ table("course_concept_stats").select(
+ "concept_name,common_misconceptions",
+ filters={"course_id": f"eq.{course_id}"},
+ order="updated_at.desc",
+ limit=20,
+ )
+ or []
+ )
+ except Exception:
+ logger.exception(
+ "read_misconceptions_for_course failed for course=%s",
+ course_id,
+ )
+ return []
+
+ rows = await asyncio.to_thread(_fetch)
+ out: list[Misconception] = []
+ seen: set[str] = set()
+ for r in rows:
+ concept = r.get("concept_name") or None
+ for m in r.get("common_misconceptions") or []:
+ text = (m or "").strip() if isinstance(m, str) else ""
+ if not text:
+ continue
+ key = text.lower()
+ if key in seen:
+ continue
+ seen.add(key)
+ out.append(Misconception(text=text, related_concept=concept))
+ return out
+
+
+async def read_misconceptions_for_course_tool(
+ ctx: RunContext[SaplingDeps],
+) -> list[Misconception]:
+ """Pydantic AI tool wrapper. Reads from ctx.deps."""
+ return await read_misconceptions_for_course(ctx.deps.course_id)

backend/models/__init__.py

@@ -44,6 +44,12 @@ class GenerateQuizBody(BaseModel):
  num_questions: int = 5
  difficulty: str = "medium"
  use_shared_context: bool = True
+ # Mirrors the Learn-route fast/smart toggle so quiz generation has
+ # the same per-request model-quality choice as chat tutoring.
+ # "fast" → gemini-2.5-flash, "smart" → gemini-2.5-pro. None falls
+ # through to whatever SAPLING_MODEL_QUIZ resolves to (default
+ # gemini-2.5-flash-lite per ADR 0008).
+ model_pref: Optional[Literal["fast", "smart"]] = None
 
 
 class AnswerItem(BaseModel):