feat(quiz): adaptive iteration — spaced repetition + history + difficulty

backend/agents/quiz.py

@@ -24,6 +24,7 @@
  read_concepts_for_user_tool,
  read_misconceptions_for_course_tool,
 )
+from agents.tools.quiz_history import read_recent_quiz_attempts_tool
 
 
 # Difficulty + question type are Literals so Gemini's enum constraint
@@ -69,31 +70,63 @@ class Quiz(BaseModel):
 _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"
+ "mastery on, OR address a class-level misconception you've seen, "
+ "OR revive a concept the student hasn't reviewed in a while.\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"
+ " weakest first). Each concept also carries `last_reviewed_at`, "
+ " which you use for spaced repetition (see rules below).\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"
+ "3. Call `read_recent_quiz_attempts(concept_node_id)` for the "
+ " target concept_node_id given in the user message. The "
+ " `summary` is a digest of past mistakes the student has made "
+ " on this concept — mine it for distractor inspiration. The "
+ " `recent_attempts` list (newest first) drives adaptive "
+ " difficulty (see rules below).\n"
+ "4. Compose `Quiz.questions` so the WEAKEST and STALEST concepts "
+ " get the most questions, AND each item's `concept` field "
+ " exactly matches a concept_name returned by tool 1.\n\n"
+ "Concept-selection rules (combine all three signals):\n"
+ "- Bias question count toward the lowest-mastery concepts (the "
+ " weakest first in the tool 1 return).\n"
+ "- SPACED REPETITION: also surface concepts whose "
+ " `last_reviewed_at` is older than ~7 days, even if their "
+ " mastery is mid-tier — they're due for review and decay over "
+ " time. Concepts with `last_reviewed_at = null` are unreviewed; "
+ " treat them as stale.\n"
+ "- Don't drop high-mastery, recently-reviewed concepts entirely; "
+ " include 1 question on a strong-and-fresh concept to keep the "
+ " quiz from feeling punishing.\n\n"
+ "Adaptive-difficulty rules (use `recent_attempts.accuracy`):\n"
+ "- If the most recent 2-3 attempts on this concept averaged < "
+ " 0.5 accuracy, drop the difficulty mix one step from what the "
+ " user asked (hard -> medium, medium -> easy, easy stays easy). "
+ " The student is struggling; keep them on track.\n"
+ "- If the most recent 3 attempts all scored >= 0.8, you may "
+ " include 1-2 questions one step harder than the requested "
+ " difficulty to push them.\n"
+ "- If `recent_attempts` is empty (first attempt), honor the "
+ " user-requested difficulty exactly.\n"
+ "- Never override the user-requested difficulty by more than one "
+ " step in either direction. Stay close to what they asked for.\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"
+ " noise. Combine signals from `read_misconceptions_for_course` "
+ " (class-wide) and `read_recent_quiz_attempts.summary` "
+ " (this student's prior errors) when writing them.\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."
+ "- difficulty: align with the student's mastery on the concept "
+ " AND the adaptive-difficulty rules above.\n\n"
+ "Honor the requested num_questions. Don't invent concepts the "
+ "student doesn't have."
 )
 _PROMPT_HASH = hashlib.sha256(_SYSTEM_PROMPT.encode("utf-8")).hexdigest()[:12]
 
@@ -107,5 +140,6 @@ class Quiz(BaseModel):
  tools=[
  read_concepts_for_user_tool,
  read_misconceptions_for_course_tool,
+ read_recent_quiz_attempts_tool,
  ],
 )

backend/agents/tools/quiz_history.py

@@ -0,0 +1,234 @@
+"""Quiz-history read tool for the quiz agent.
+
+Surfaces what the student previously got wrong on a concept and how
+their last few attempts scored. The agent uses this for two things:
+
+1. Targeting — write distractors that mirror the student's prior
+ mistakes (the LLM-generated `summary` from `quiz_context`
+ captures patterns rolled up across past attempts).
+2. Adaptive difficulty — read the last few `quiz_attempts` rows and
+ step difficulty down when the student has been struggling, up
+ when they've been crushing it.
+
+The pure async function is callable from routes/tests; the *_tool
+wrapper registers 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__)
+
+
+# How many past attempts the agent gets to see. 5 is enough to spot a
+# trend without flooding the prompt with state. Older attempts are
+# already rolled into `summary` by the post-quiz context update job.
+_RECENT_ATTEMPTS_LIMIT = 5
+
+
+class RecentQuizAttempt(BaseModel):
+ """One past attempt's headline numbers."""
+
+ score: int = Field(ge=0)
+ total: int = Field(ge=0)
+ difficulty: str
+ completed_at: str | None = None
+ accuracy: float = Field(ge=0.0, le=1.0)
+
+
+class QuizHistory(BaseModel):
+ """The agent's view of a student's history on one concept."""
+
+ # LLM-generated digest of past quiz mistakes/patterns for this
+ # (user, concept). Populated by the background context-update job
+ # in routes/quiz.py:submit_quiz. May be None on a first attempt.
+ summary: str | None = None
+ # Most recent attempts, newest first. Empty on first attempt.
+ recent_attempts: list[RecentQuizAttempt] = Field(default_factory=list)
+
+
+def _coerce_summary(ctx: Any) -> str | None:
+ """quiz_context.context_json is free-form (whatever the post-submit
+ LLM produced). Different prompt versions have stored either a flat
+ string or a small dict. Coerce to a single string the agent can
+ reason over, or None if there's nothing useful."""
+ if not ctx:
+ return None
+ if isinstance(ctx, str):
+ text = ctx.strip()
+ return text or None
+ if isinstance(ctx, dict):
+ # Common shapes: {"summary": "..."}, {"notes": "..."},
+ # {"misconceptions": [...], "weak_areas": [...]}.
+ for key in ("summary", "notes", "context", "digest"):
+ v = ctx.get(key)
+ if isinstance(v, str) and v.strip():
+ return v.strip()
+ # Fall back to flattening list-of-strings entries so the agent
+ # at least sees the misconceptions/weak_areas the prior job
+ # extracted, even when no top-level summary string exists.
+ parts: list[str] = []
+ for key in ("misconceptions", "weak_areas", "common_errors"):
+ for item in ctx.get(key) or []:
+ if isinstance(item, str) and item.strip():
+ parts.append(f"- {item.strip()}")
+ return "\n".join(parts) or None
+ return None
+
+
+async def read_recent_quiz_attempts(
+ user_id: str,
+ concept_node_id: str,
+) -> QuizHistory:
+ """Return the agent's view of a student's history on one concept.
+
+ Reads two sources:
+
+ - `quiz_context` (one row per (user, concept)): the rolling
+ LLM-generated digest of what the student has been getting
+ wrong. This is the same blob legacy `routes/quiz.py` used to
+ stuff into the prompt template.
+ - `quiz_attempts` (one row per attempt, filtered to completed
+ attempts): the last N completed attempts, newest first, with
+ accuracy precomputed so the agent doesn't have to.
+
+ Wraps the sync Supabase reads in `asyncio.to_thread` so we don't
+ block the event loop. Failures degrade silently — the agent can
+ still generate a quiz without history (just less adaptive).
+ """
+
+ def _fetch_summary() -> Any:
+ try:
+ rows = table("quiz_context").select(
+ "context_json",
+ filters={
+ "user_id": f"eq.{user_id}",
+ "concept_node_id": f"eq.{concept_node_id}",
+ },
+ limit=1,
+ )
+ return rows[0]["context_json"] if rows else None
+ except Exception:
+ logger.exception(
+ "read_recent_quiz_attempts: quiz_context fetch failed "
+ "user=%s concept=%s",
+ user_id,
+ concept_node_id,
+ )
+ return None
+
+ def _fetch_attempts() -> list[dict[str, Any]]:
+ try:
+ return (
+ table("quiz_attempts").select(
+ "score,total,difficulty,completed_at",
+ filters={
+ "user_id": f"eq.{user_id}",
+ "concept_node_id": f"eq.{concept_node_id}",
+ # Only count completed attempts. PostgREST `not.is.null`
+ # filters out rows where completed_at is NULL, which is
+ # how `routes/quiz.py:generate_quiz` marks an in-flight
+ # attempt before submission.
+ "completed_at": "not.is.null",
+ },
+ order="completed_at.desc",
+ limit=_RECENT_ATTEMPTS_LIMIT,
+ )
+ or []
+ )
+ except Exception:
+ logger.exception(
+ "read_recent_quiz_attempts: quiz_attempts fetch failed "
+ "user=%s concept=%s",
+ user_id,
+ concept_node_id,
+ )
+ return []
+
+ summary_raw, attempt_rows = await asyncio.gather(
+ asyncio.to_thread(_fetch_summary),
+ asyncio.to_thread(_fetch_attempts),
+ )
+
+ attempts: list[RecentQuizAttempt] = []
+ for r in attempt_rows:
+ raw_score = r.get("score")
+ raw_total = r.get("total")
+ if raw_score is None or raw_total is None:
+ # `submit_quiz` writes score+total atomically, so a row with
+ # `completed_at IS NOT NULL` but a null score/total is
+ # corruption (or an out-of-band edit). Drop it rather than
+ # coercing to 0/0 — feeding the LLM a bogus 0% accuracy
+ # could trigger a spurious adaptive downshift.
+ logger.warning(
+ "read_recent_quiz_attempts: dropping row with null "
+ "score/total (score=%r, total=%r) user=%s concept=%s",
+ raw_score,
+ raw_total,
+ user_id,
+ concept_node_id,
+ )
+ continue
+ try:
+ score = int(raw_score)
+ total = int(raw_total)
+ except (TypeError, ValueError):
+ continue
+ if total <= 0:
+ # Skip rows that look incomplete — accuracy is undefined and
+ # the agent shouldn't have to guess.
+ continue
+ if score < 0 or score > total:
+ # Corrupt row (score outside [0, total]). Drop entirely
+ # rather than passing impossible numbers to the LLM —
+ # `score=7, total=5` would prompt the agent to wonder
+ # whether to trust the data at all.
+ logger.warning(
+ "read_recent_quiz_attempts: dropping corrupt row "
+ "(score=%d outside [0, total=%d]) user=%s concept=%s",
+ score,
+ total,
+ user_id,
+ concept_node_id,
+ )
+ continue
+ accuracy = score / total
+ attempts.append(
+ RecentQuizAttempt(
+ score=score,
+ total=total,
+ difficulty=str(r.get("difficulty") or ""),
+ completed_at=r.get("completed_at"),
+ accuracy=round(accuracy, 4),
+ )
+ )
+
+ return QuizHistory(
+ summary=_coerce_summary(summary_raw),
+ recent_attempts=attempts,
+ )
+
+
+async def read_recent_quiz_attempts_tool(
+ ctx: RunContext[SaplingDeps],
+ concept_node_id: str,
+) -> QuizHistory:
+ """Returns this student's history on one concept: a `summary`
+ string digesting their prior mistakes (mine for distractor
+ inspiration) and `recent_attempts` — the last 5 completed quiz
+ attempts on this concept, newest first, with `accuracy` precomputed
+ so you can apply the adaptive-difficulty rule directly. Empty
+ history on first attempt. Pass the `concept_node_id` from the
+ user message; user identity is taken from context.
+ """
+ # user_id comes from ctx.deps so a tool call can't cross users.
+ return await read_recent_quiz_attempts(ctx.deps.user_id, concept_node_id)

backend/routes/quiz.py

@@ -166,11 +166,15 @@ async def _quiz_via_agent(
  supabase=None,
  request_id=request_id,
  )
+ # Keep this message routing-only; the workflow + adaptive rules
+ # live in the system prompt. We just hand the agent the inputs it
+ # needs and trust the prompt to drive tool calls.
  user_message = (
  f"Generate {num_questions} {difficulty} questions for the student. "
- f"The target concept is '{concept_name}' (concept_node_id={concept_node_id}). "
- f"Call read_concepts_for_user to find the student's weakest concepts in this course "
- f"and bias the question mix toward those."
+ f"The target concept is '{concept_name}' "
+ f"(concept_node_id={concept_node_id}). Follow the workflow in your "
+ f"system prompt; pass concept_node_id='{concept_node_id}' to "
+ f"read_recent_quiz_attempts."
  )
  if use_shared_context:
  user_message += (

backend/agents/quiz.py

@@ -24,6 +24,7 @@
  read_concepts_for_user_tool,
  read_misconceptions_for_course_tool,
 )
+from agents.tools.quiz_history import read_recent_quiz_attempts_tool
 
 
 # Difficulty + question type are Literals so Gemini's enum constraint
@@ -69,31 +70,63 @@ class Quiz(BaseModel):
 _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"
+ "mastery on, OR address a class-level misconception you've seen, "
+ "OR revive a concept the student hasn't reviewed in a while.\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"
+ " weakest first). Each concept also carries `last_reviewed_at`, "
+ " which you use for spaced repetition (see rules below).\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"
+ "3. Call `read_recent_quiz_attempts(concept_node_id)` for the "
+ " target concept_node_id given in the user message. The "
+ " `summary` is a digest of past mistakes the student has made "
+ " on this concept — mine it for distractor inspiration. The "
+ " `recent_attempts` list (newest first) drives adaptive "
+ " difficulty (see rules below).\n"
+ "4. Compose `Quiz.questions` so the WEAKEST and STALEST concepts "
+ " get the most questions, AND each item's `concept` field "
+ " exactly matches a concept_name returned by tool 1.\n\n"
+ "Concept-selection rules (combine all three signals):\n"
+ "- Bias question count toward the lowest-mastery concepts (the "
+ " weakest first in the tool 1 return).\n"
+ "- SPACED REPETITION: also surface concepts whose "
+ " `last_reviewed_at` is older than ~7 days, even if their "
+ " mastery is mid-tier — they're due for review and decay over "
+ " time. Concepts with `last_reviewed_at = null` are unreviewed; "
+ " treat them as stale.\n"
+ "- Don't drop high-mastery, recently-reviewed concepts entirely; "
+ " include 1 question on a strong-and-fresh concept to keep the "
+ " quiz from feeling punishing.\n\n"
+ "Adaptive-difficulty rules (use `recent_attempts.accuracy`):\n"
+ "- If the most recent 2-3 attempts on this concept averaged < "
+ " 0.5 accuracy, drop the difficulty mix one step from what the "
+ " user asked (hard -> medium, medium -> easy, easy stays easy). "
+ " The student is struggling; keep them on track.\n"
+ "- If the most recent 3 attempts all scored >= 0.8, you may "
+ " include 1-2 questions one step harder than the requested "
+ " difficulty to push them.\n"
+ "- If `recent_attempts` is empty (first attempt), honor the "
+ " user-requested difficulty exactly.\n"
+ "- Never override the user-requested difficulty by more than one "
+ " step in either direction. Stay close to what they asked for.\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"
+ " noise. Combine signals from `read_misconceptions_for_course` "
+ " (class-wide) and `read_recent_quiz_attempts.summary` "
+ " (this student's prior errors) when writing them.\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."
+ "- difficulty: align with the student's mastery on the concept "
+ " AND the adaptive-difficulty rules above.\n\n"
+ "Honor the requested num_questions. Don't invent concepts the "
+ "student doesn't have."
 )
 _PROMPT_HASH = hashlib.sha256(_SYSTEM_PROMPT.encode("utf-8")).hexdigest()[:12]
 
@@ -107,5 +140,6 @@ class Quiz(BaseModel):
  tools=[
  read_concepts_for_user_tool,
  read_misconceptions_for_course_tool,
+ read_recent_quiz_attempts_tool,
  ],
 )

backend/agents/tools/quiz_history.py

@@ -0,0 +1,234 @@
+"""Quiz-history read tool for the quiz agent.
+
+Surfaces what the student previously got wrong on a concept and how
+their last few attempts scored. The agent uses this for two things:
+
+1. Targeting — write distractors that mirror the student's prior
+ mistakes (the LLM-generated `summary` from `quiz_context`
+ captures patterns rolled up across past attempts).
+2. Adaptive difficulty — read the last few `quiz_attempts` rows and
+ step difficulty down when the student has been struggling, up
+ when they've been crushing it.
+
+The pure async function is callable from routes/tests; the *_tool
+wrapper registers 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__)
+
+
+# How many past attempts the agent gets to see. 5 is enough to spot a
+# trend without flooding the prompt with state. Older attempts are
+# already rolled into `summary` by the post-quiz context update job.
+_RECENT_ATTEMPTS_LIMIT = 5
+
+
+class RecentQuizAttempt(BaseModel):
+ """One past attempt's headline numbers."""
+
+ score: int = Field(ge=0)
+ total: int = Field(ge=0)
+ difficulty: str
+ completed_at: str | None = None
+ accuracy: float = Field(ge=0.0, le=1.0)
+
+
+class QuizHistory(BaseModel):
+ """The agent's view of a student's history on one concept."""
+
+ # LLM-generated digest of past quiz mistakes/patterns for this
+ # (user, concept). Populated by the background context-update job
+ # in routes/quiz.py:submit_quiz. May be None on a first attempt.
+ summary: str | None = None
+ # Most recent attempts, newest first. Empty on first attempt.
+ recent_attempts: list[RecentQuizAttempt] = Field(default_factory=list)
+
+
+def _coerce_summary(ctx: Any) -> str | None:
+ """quiz_context.context_json is free-form (whatever the post-submit
+ LLM produced). Different prompt versions have stored either a flat
+ string or a small dict. Coerce to a single string the agent can
+ reason over, or None if there's nothing useful."""
+ if not ctx:
+ return None
+ if isinstance(ctx, str):
+ text = ctx.strip()
+ return text or None
+ if isinstance(ctx, dict):
+ # Common shapes: {"summary": "..."}, {"notes": "..."},
+ # {"misconceptions": [...], "weak_areas": [...]}.
+ for key in ("summary", "notes", "context", "digest"):
+ v = ctx.get(key)
+ if isinstance(v, str) and v.strip():
+ return v.strip()
+ # Fall back to flattening list-of-strings entries so the agent
+ # at least sees the misconceptions/weak_areas the prior job
+ # extracted, even when no top-level summary string exists.
+ parts: list[str] = []
+ for key in ("misconceptions", "weak_areas", "common_errors"):
+ for item in ctx.get(key) or []:
+ if isinstance(item, str) and item.strip():
+ parts.append(f"- {item.strip()}")
+ return "\n".join(parts) or None
+ return None
+
+
+async def read_recent_quiz_attempts(
+ user_id: str,
+ concept_node_id: str,
+) -> QuizHistory:
+ """Return the agent's view of a student's history on one concept.
+
+ Reads two sources:
+
+ - `quiz_context` (one row per (user, concept)): the rolling
+ LLM-generated digest of what the student has been getting
+ wrong. This is the same blob legacy `routes/quiz.py` used to
+ stuff into the prompt template.
+ - `quiz_attempts` (one row per attempt, filtered to completed
+ attempts): the last N completed attempts, newest first, with
+ accuracy precomputed so the agent doesn't have to.
+
+ Wraps the sync Supabase reads in `asyncio.to_thread` so we don't
+ block the event loop. Failures degrade silently — the agent can
+ still generate a quiz without history (just less adaptive).
+ """
+
+ def _fetch_summary() -> Any:
+ try:
+ rows = table("quiz_context").select(
+ "context_json",
+ filters={
+ "user_id": f"eq.{user_id}",
+ "concept_node_id": f"eq.{concept_node_id}",
+ },
+ limit=1,
+ )
+ return rows[0]["context_json"] if rows else None
+ except Exception:
+ logger.exception(
+ "read_recent_quiz_attempts: quiz_context fetch failed "
+ "user=%s concept=%s",
+ user_id,
+ concept_node_id,
+ )
+ return None
+
+ def _fetch_attempts() -> list[dict[str, Any]]:
+ try:
+ return (
+ table("quiz_attempts").select(
+ "score,total,difficulty,completed_at",
+ filters={
+ "user_id": f"eq.{user_id}",
+ "concept_node_id": f"eq.{concept_node_id}",
+ # Only count completed attempts. PostgREST `not.is.null`
+ # filters out rows where completed_at is NULL, which is
+ # how `routes/quiz.py:generate_quiz` marks an in-flight
+ # attempt before submission.
+ "completed_at": "not.is.null",
+ },
+ order="completed_at.desc",
+ limit=_RECENT_ATTEMPTS_LIMIT,
+ )
+ or []
+ )
+ except Exception:
+ logger.exception(
+ "read_recent_quiz_attempts: quiz_attempts fetch failed "
+ "user=%s concept=%s",
+ user_id,
+ concept_node_id,
+ )
+ return []
+
+ summary_raw, attempt_rows = await asyncio.gather(
+ asyncio.to_thread(_fetch_summary),
+ asyncio.to_thread(_fetch_attempts),
+ )
+
+ attempts: list[RecentQuizAttempt] = []
+ for r in attempt_rows:
+ raw_score = r.get("score")
+ raw_total = r.get("total")
+ if raw_score is None or raw_total is None:
+ # `submit_quiz` writes score+total atomically, so a row with
+ # `completed_at IS NOT NULL` but a null score/total is
+ # corruption (or an out-of-band edit). Drop it rather than
+ # coercing to 0/0 — feeding the LLM a bogus 0% accuracy
+ # could trigger a spurious adaptive downshift.
+ logger.warning(
+ "read_recent_quiz_attempts: dropping row with null "
+ "score/total (score=%r, total=%r) user=%s concept=%s",
+ raw_score,
+ raw_total,
+ user_id,
+ concept_node_id,
+ )
+ continue
+ try:
+ score = int(raw_score)
+ total = int(raw_total)
+ except (TypeError, ValueError):
+ continue
+ if total <= 0:
+ # Skip rows that look incomplete — accuracy is undefined and
+ # the agent shouldn't have to guess.
+ continue
+ if score < 0 or score > total:
+ # Corrupt row (score outside [0, total]). Drop entirely
+ # rather than passing impossible numbers to the LLM —
+ # `score=7, total=5` would prompt the agent to wonder
+ # whether to trust the data at all.
+ logger.warning(
+ "read_recent_quiz_attempts: dropping corrupt row "
+ "(score=%d outside [0, total=%d]) user=%s concept=%s",
+ score,
+ total,
+ user_id,
+ concept_node_id,
+ )
+ continue
+ accuracy = score / total
+ attempts.append(
+ RecentQuizAttempt(
+ score=score,
+ total=total,
+ difficulty=str(r.get("difficulty") or ""),
+ completed_at=r.get("completed_at"),
+ accuracy=round(accuracy, 4),
+ )
+ )
+
+ return QuizHistory(
+ summary=_coerce_summary(summary_raw),
+ recent_attempts=attempts,
+ )
+
+
+async def read_recent_quiz_attempts_tool(
+ ctx: RunContext[SaplingDeps],
+ concept_node_id: str,
+) -> QuizHistory:
+ """Returns this student's history on one concept: a `summary`
+ string digesting their prior mistakes (mine for distractor
+ inspiration) and `recent_attempts` — the last 5 completed quiz
+ attempts on this concept, newest first, with `accuracy` precomputed
+ so you can apply the adaptive-difficulty rule directly. Empty
+ history on first attempt. Pass the `concept_node_id` from the
+ user message; user identity is taken from context.
+ """
+ # user_id comes from ctx.deps so a tool call can't cross users.
+ return await read_recent_quiz_attempts(ctx.deps.user_id, concept_node_id)

backend/routes/quiz.py

@@ -166,11 +166,15 @@ async def _quiz_via_agent(
  supabase=None,
  request_id=request_id,
  )
+ # Keep this message routing-only; the workflow + adaptive rules
+ # live in the system prompt. We just hand the agent the inputs it
+ # needs and trust the prompt to drive tool calls.
  user_message = (
  f"Generate {num_questions} {difficulty} questions for the student. "
- f"The target concept is '{concept_name}' (concept_node_id={concept_node_id}). "
- f"Call read_concepts_for_user to find the student's weakest concepts in this course "
- f"and bias the question mix toward those."
+ f"The target concept is '{concept_name}' "
+ f"(concept_node_id={concept_node_id}). Follow the workflow in your "
+ f"system prompt; pass concept_node_id='{concept_node_id}' to "
+ f"read_recent_quiz_attempts."
  )
  if use_shared_context:
  user_message += (

backend/agents/quiz.py

@@ -24,6 +24,7 @@
  read_concepts_for_user_tool,
  read_misconceptions_for_course_tool,
 )
+from agents.tools.quiz_history import read_recent_quiz_attempts_tool
 
 
 # Difficulty + question type are Literals so Gemini's enum constraint
@@ -69,31 +70,63 @@ class Quiz(BaseModel):
 _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"
+ "mastery on, OR address a class-level misconception you've seen, "
+ "OR revive a concept the student hasn't reviewed in a while.\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"
+ " weakest first). Each concept also carries `last_reviewed_at`, "
+ " which you use for spaced repetition (see rules below).\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"
+ "3. Call `read_recent_quiz_attempts(concept_node_id)` for the "
+ " target concept_node_id given in the user message. The "
+ " `summary` is a digest of past mistakes the student has made "
+ " on this concept — mine it for distractor inspiration. The "
+ " `recent_attempts` list (newest first) drives adaptive "
+ " difficulty (see rules below).\n"
+ "4. Compose `Quiz.questions` so the WEAKEST and STALEST concepts "
+ " get the most questions, AND each item's `concept` field "
+ " exactly matches a concept_name returned by tool 1.\n\n"
+ "Concept-selection rules (combine all three signals):\n"
+ "- Bias question count toward the lowest-mastery concepts (the "
+ " weakest first in the tool 1 return).\n"
+ "- SPACED REPETITION: also surface concepts whose "
+ " `last_reviewed_at` is older than ~7 days, even if their "
+ " mastery is mid-tier — they're due for review and decay over "
+ " time. Concepts with `last_reviewed_at = null` are unreviewed; "
+ " treat them as stale.\n"
+ "- Don't drop high-mastery, recently-reviewed concepts entirely; "
+ " include 1 question on a strong-and-fresh concept to keep the "
+ " quiz from feeling punishing.\n\n"
+ "Adaptive-difficulty rules (use `recent_attempts.accuracy`):\n"
+ "- If the most recent 2-3 attempts on this concept averaged < "
+ " 0.5 accuracy, drop the difficulty mix one step from what the "
+ " user asked (hard -> medium, medium -> easy, easy stays easy). "
+ " The student is struggling; keep them on track.\n"
+ "- If the most recent 3 attempts all scored >= 0.8, you may "
+ " include 1-2 questions one step harder than the requested "
+ " difficulty to push them.\n"
+ "- If `recent_attempts` is empty (first attempt), honor the "
+ " user-requested difficulty exactly.\n"
+ "- Never override the user-requested difficulty by more than one "
+ " step in either direction. Stay close to what they asked for.\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"
+ " noise. Combine signals from `read_misconceptions_for_course` "
+ " (class-wide) and `read_recent_quiz_attempts.summary` "
+ " (this student's prior errors) when writing them.\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."
+ "- difficulty: align with the student's mastery on the concept "
+ " AND the adaptive-difficulty rules above.\n\n"
+ "Honor the requested num_questions. Don't invent concepts the "
+ "student doesn't have."
 )
 _PROMPT_HASH = hashlib.sha256(_SYSTEM_PROMPT.encode("utf-8")).hexdigest()[:12]
 
@@ -107,5 +140,6 @@ class Quiz(BaseModel):
  tools=[
  read_concepts_for_user_tool,
  read_misconceptions_for_course_tool,
+ read_recent_quiz_attempts_tool,
  ],
 )

backend/agents/tools/quiz_history.py

@@ -0,0 +1,234 @@
+"""Quiz-history read tool for the quiz agent.
+
+Surfaces what the student previously got wrong on a concept and how
+their last few attempts scored. The agent uses this for two things:
+
+1. Targeting — write distractors that mirror the student's prior
+ mistakes (the LLM-generated `summary` from `quiz_context`
+ captures patterns rolled up across past attempts).
+2. Adaptive difficulty — read the last few `quiz_attempts` rows and
+ step difficulty down when the student has been struggling, up
+ when they've been crushing it.
+
+The pure async function is callable from routes/tests; the *_tool
+wrapper registers 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__)
+
+
+# How many past attempts the agent gets to see. 5 is enough to spot a
+# trend without flooding the prompt with state. Older attempts are
+# already rolled into `summary` by the post-quiz context update job.
+_RECENT_ATTEMPTS_LIMIT = 5
+
+
+class RecentQuizAttempt(BaseModel):
+ """One past attempt's headline numbers."""
+
+ score: int = Field(ge=0)
+ total: int = Field(ge=0)
+ difficulty: str
+ completed_at: str | None = None
+ accuracy: float = Field(ge=0.0, le=1.0)
+
+
+class QuizHistory(BaseModel):
+ """The agent's view of a student's history on one concept."""
+
+ # LLM-generated digest of past quiz mistakes/patterns for this
+ # (user, concept). Populated by the background context-update job
+ # in routes/quiz.py:submit_quiz. May be None on a first attempt.
+ summary: str | None = None
+ # Most recent attempts, newest first. Empty on first attempt.
+ recent_attempts: list[RecentQuizAttempt] = Field(default_factory=list)
+
+
+def _coerce_summary(ctx: Any) -> str | None:
+ """quiz_context.context_json is free-form (whatever the post-submit
+ LLM produced). Different prompt versions have stored either a flat
+ string or a small dict. Coerce to a single string the agent can
+ reason over, or None if there's nothing useful."""
+ if not ctx:
+ return None
+ if isinstance(ctx, str):
+ text = ctx.strip()
+ return text or None
+ if isinstance(ctx, dict):
+ # Common shapes: {"summary": "..."}, {"notes": "..."},
+ # {"misconceptions": [...], "weak_areas": [...]}.
+ for key in ("summary", "notes", "context", "digest"):
+ v = ctx.get(key)
+ if isinstance(v, str) and v.strip():
+ return v.strip()
+ # Fall back to flattening list-of-strings entries so the agent
+ # at least sees the misconceptions/weak_areas the prior job
+ # extracted, even when no top-level summary string exists.
+ parts: list[str] = []
+ for key in ("misconceptions", "weak_areas", "common_errors"):
+ for item in ctx.get(key) or []:
+ if isinstance(item, str) and item.strip():
+ parts.append(f"- {item.strip()}")
+ return "\n".join(parts) or None
+ return None
+
+
+async def read_recent_quiz_attempts(
+ user_id: str,
+ concept_node_id: str,
+) -> QuizHistory:
+ """Return the agent's view of a student's history on one concept.
+
+ Reads two sources:
+
+ - `quiz_context` (one row per (user, concept)): the rolling
+ LLM-generated digest of what the student has been getting
+ wrong. This is the same blob legacy `routes/quiz.py` used to
+ stuff into the prompt template.
+ - `quiz_attempts` (one row per attempt, filtered to completed
+ attempts): the last N completed attempts, newest first, with
+ accuracy precomputed so the agent doesn't have to.
+
+ Wraps the sync Supabase reads in `asyncio.to_thread` so we don't
+ block the event loop. Failures degrade silently — the agent can
+ still generate a quiz without history (just less adaptive).
+ """
+
+ def _fetch_summary() -> Any:
+ try:
+ rows = table("quiz_context").select(
+ "context_json",
+ filters={
+ "user_id": f"eq.{user_id}",
+ "concept_node_id": f"eq.{concept_node_id}",
+ },
+ limit=1,
+ )
+ return rows[0]["context_json"] if rows else None
+ except Exception:
+ logger.exception(
+ "read_recent_quiz_attempts: quiz_context fetch failed "
+ "user=%s concept=%s",
+ user_id,
+ concept_node_id,
+ )
+ return None
+
+ def _fetch_attempts() -> list[dict[str, Any]]:
+ try:
+ return (
+ table("quiz_attempts").select(
+ "score,total,difficulty,completed_at",
+ filters={
+ "user_id": f"eq.{user_id}",
+ "concept_node_id": f"eq.{concept_node_id}",
+ # Only count completed attempts. PostgREST `not.is.null`
+ # filters out rows where completed_at is NULL, which is
+ # how `routes/quiz.py:generate_quiz` marks an in-flight
+ # attempt before submission.
+ "completed_at": "not.is.null",
+ },
+ order="completed_at.desc",
+ limit=_RECENT_ATTEMPTS_LIMIT,
+ )
+ or []
+ )
+ except Exception:
+ logger.exception(
+ "read_recent_quiz_attempts: quiz_attempts fetch failed "
+ "user=%s concept=%s",
+ user_id,
+ concept_node_id,
+ )
+ return []
+
+ summary_raw, attempt_rows = await asyncio.gather(
+ asyncio.to_thread(_fetch_summary),
+ asyncio.to_thread(_fetch_attempts),
+ )
+
+ attempts: list[RecentQuizAttempt] = []
+ for r in attempt_rows:
+ raw_score = r.get("score")
+ raw_total = r.get("total")
+ if raw_score is None or raw_total is None:
+ # `submit_quiz` writes score+total atomically, so a row with
+ # `completed_at IS NOT NULL` but a null score/total is
+ # corruption (or an out-of-band edit). Drop it rather than
+ # coercing to 0/0 — feeding the LLM a bogus 0% accuracy
+ # could trigger a spurious adaptive downshift.
+ logger.warning(
+ "read_recent_quiz_attempts: dropping row with null "
+ "score/total (score=%r, total=%r) user=%s concept=%s",
+ raw_score,
+ raw_total,
+ user_id,
+ concept_node_id,
+ )
+ continue
+ try:
+ score = int(raw_score)
+ total = int(raw_total)
+ except (TypeError, ValueError):
+ continue
+ if total <= 0:
+ # Skip rows that look incomplete — accuracy is undefined and
+ # the agent shouldn't have to guess.
+ continue
+ if score < 0 or score > total:
+ # Corrupt row (score outside [0, total]). Drop entirely
+ # rather than passing impossible numbers to the LLM —
+ # `score=7, total=5` would prompt the agent to wonder
+ # whether to trust the data at all.
+ logger.warning(
+ "read_recent_quiz_attempts: dropping corrupt row "
+ "(score=%d outside [0, total=%d]) user=%s concept=%s",
+ score,
+ total,
+ user_id,
+ concept_node_id,
+ )
+ continue
+ accuracy = score / total
+ attempts.append(
+ RecentQuizAttempt(
+ score=score,
+ total=total,
+ difficulty=str(r.get("difficulty") or ""),
+ completed_at=r.get("completed_at"),
+ accuracy=round(accuracy, 4),
+ )
+ )
+
+ return QuizHistory(
+ summary=_coerce_summary(summary_raw),
+ recent_attempts=attempts,
+ )
+
+
+async def read_recent_quiz_attempts_tool(
+ ctx: RunContext[SaplingDeps],
+ concept_node_id: str,
+) -> QuizHistory:
+ """Returns this student's history on one concept: a `summary`
+ string digesting their prior mistakes (mine for distractor
+ inspiration) and `recent_attempts` — the last 5 completed quiz
+ attempts on this concept, newest first, with `accuracy` precomputed
+ so you can apply the adaptive-difficulty rule directly. Empty
+ history on first attempt. Pass the `concept_node_id` from the
+ user message; user identity is taken from context.
+ """
+ # user_id comes from ctx.deps so a tool call can't cross users.
+ return await read_recent_quiz_attempts(ctx.deps.user_id, concept_node_id)

backend/routes/quiz.py

@@ -166,11 +166,15 @@ async def _quiz_via_agent(
  supabase=None,
  request_id=request_id,
  )
+ # Keep this message routing-only; the workflow + adaptive rules
+ # live in the system prompt. We just hand the agent the inputs it
+ # needs and trust the prompt to drive tool calls.
  user_message = (
  f"Generate {num_questions} {difficulty} questions for the student. "
- f"The target concept is '{concept_name}' (concept_node_id={concept_node_id}). "
- f"Call read_concepts_for_user to find the student's weakest concepts in this course "
- f"and bias the question mix toward those."
+ f"The target concept is '{concept_name}' "
+ f"(concept_node_id={concept_node_id}). Follow the workflow in your "
+ f"system prompt; pass concept_node_id='{concept_node_id}' to "
+ f"read_recent_quiz_attempts."
  )
  if use_shared_context:
  user_message += (

backend/agents/quiz.py

@@ -24,6 +24,7 @@
  read_concepts_for_user_tool,
  read_misconceptions_for_course_tool,
 )
+from agents.tools.quiz_history import read_recent_quiz_attempts_tool
 
 
 # Difficulty + question type are Literals so Gemini's enum constraint
@@ -69,31 +70,63 @@ class Quiz(BaseModel):
 _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"
+ "mastery on, OR address a class-level misconception you've seen, "
+ "OR revive a concept the student hasn't reviewed in a while.\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"
+ " weakest first). Each concept also carries `last_reviewed_at`, "
+ " which you use for spaced repetition (see rules below).\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"
+ "3. Call `read_recent_quiz_attempts(concept_node_id)` for the "
+ " target concept_node_id given in the user message. The "
+ " `summary` is a digest of past mistakes the student has made "
+ " on this concept — mine it for distractor inspiration. The "
+ " `recent_attempts` list (newest first) drives adaptive "
+ " difficulty (see rules below).\n"
+ "4. Compose `Quiz.questions` so the WEAKEST and STALEST concepts "
+ " get the most questions, AND each item's `concept` field "
+ " exactly matches a concept_name returned by tool 1.\n\n"
+ "Concept-selection rules (combine all three signals):\n"
+ "- Bias question count toward the lowest-mastery concepts (the "
+ " weakest first in the tool 1 return).\n"
+ "- SPACED REPETITION: also surface concepts whose "
+ " `last_reviewed_at` is older than ~7 days, even if their "
+ " mastery is mid-tier — they're due for review and decay over "
+ " time. Concepts with `last_reviewed_at = null` are unreviewed; "
+ " treat them as stale.\n"
+ "- Don't drop high-mastery, recently-reviewed concepts entirely; "
+ " include 1 question on a strong-and-fresh concept to keep the "
+ " quiz from feeling punishing.\n\n"
+ "Adaptive-difficulty rules (use `recent_attempts.accuracy`):\n"
+ "- If the most recent 2-3 attempts on this concept averaged < "
+ " 0.5 accuracy, drop the difficulty mix one step from what the "
+ " user asked (hard -> medium, medium -> easy, easy stays easy). "
+ " The student is struggling; keep them on track.\n"
+ "- If the most recent 3 attempts all scored >= 0.8, you may "
+ " include 1-2 questions one step harder than the requested "
+ " difficulty to push them.\n"
+ "- If `recent_attempts` is empty (first attempt), honor the "
+ " user-requested difficulty exactly.\n"
+ "- Never override the user-requested difficulty by more than one "
+ " step in either direction. Stay close to what they asked for.\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"
+ " noise. Combine signals from `read_misconceptions_for_course` "
+ " (class-wide) and `read_recent_quiz_attempts.summary` "
+ " (this student's prior errors) when writing them.\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."
+ "- difficulty: align with the student's mastery on the concept "
+ " AND the adaptive-difficulty rules above.\n\n"
+ "Honor the requested num_questions. Don't invent concepts the "
+ "student doesn't have."
 )
 _PROMPT_HASH = hashlib.sha256(_SYSTEM_PROMPT.encode("utf-8")).hexdigest()[:12]
 
@@ -107,5 +140,6 @@ class Quiz(BaseModel):
  tools=[
  read_concepts_for_user_tool,
  read_misconceptions_for_course_tool,
+ read_recent_quiz_attempts_tool,
  ],
 )

backend/agents/tools/quiz_history.py

@@ -0,0 +1,234 @@
+"""Quiz-history read tool for the quiz agent.
+
+Surfaces what the student previously got wrong on a concept and how
+their last few attempts scored. The agent uses this for two things:
+
+1. Targeting — write distractors that mirror the student's prior
+ mistakes (the LLM-generated `summary` from `quiz_context`
+ captures patterns rolled up across past attempts).
+2. Adaptive difficulty — read the last few `quiz_attempts` rows and
+ step difficulty down when the student has been struggling, up
+ when they've been crushing it.
+
+The pure async function is callable from routes/tests; the *_tool
+wrapper registers 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__)
+
+
+# How many past attempts the agent gets to see. 5 is enough to spot a
+# trend without flooding the prompt with state. Older attempts are
+# already rolled into `summary` by the post-quiz context update job.
+_RECENT_ATTEMPTS_LIMIT = 5
+
+
+class RecentQuizAttempt(BaseModel):
+ """One past attempt's headline numbers."""
+
+ score: int = Field(ge=0)
+ total: int = Field(ge=0)
+ difficulty: str
+ completed_at: str | None = None
+ accuracy: float = Field(ge=0.0, le=1.0)
+
+
+class QuizHistory(BaseModel):
+ """The agent's view of a student's history on one concept."""
+
+ # LLM-generated digest of past quiz mistakes/patterns for this
+ # (user, concept). Populated by the background context-update job
+ # in routes/quiz.py:submit_quiz. May be None on a first attempt.
+ summary: str | None = None
+ # Most recent attempts, newest first. Empty on first attempt.
+ recent_attempts: list[RecentQuizAttempt] = Field(default_factory=list)
+
+
+def _coerce_summary(ctx: Any) -> str | None:
+ """quiz_context.context_json is free-form (whatever the post-submit
+ LLM produced). Different prompt versions have stored either a flat
+ string or a small dict. Coerce to a single string the agent can
+ reason over, or None if there's nothing useful."""
+ if not ctx:
+ return None
+ if isinstance(ctx, str):
+ text = ctx.strip()
+ return text or None
+ if isinstance(ctx, dict):
+ # Common shapes: {"summary": "..."}, {"notes": "..."},
+ # {"misconceptions": [...], "weak_areas": [...]}.
+ for key in ("summary", "notes", "context", "digest"):
+ v = ctx.get(key)
+ if isinstance(v, str) and v.strip():
+ return v.strip()
+ # Fall back to flattening list-of-strings entries so the agent
+ # at least sees the misconceptions/weak_areas the prior job
+ # extracted, even when no top-level summary string exists.
+ parts: list[str] = []
+ for key in ("misconceptions", "weak_areas", "common_errors"):
+ for item in ctx.get(key) or []:
+ if isinstance(item, str) and item.strip():
+ parts.append(f"- {item.strip()}")
+ return "\n".join(parts) or None
+ return None
+
+
+async def read_recent_quiz_attempts(
+ user_id: str,
+ concept_node_id: str,
+) -> QuizHistory:
+ """Return the agent's view of a student's history on one concept.
+
+ Reads two sources:
+
+ - `quiz_context` (one row per (user, concept)): the rolling
+ LLM-generated digest of what the student has been getting
+ wrong. This is the same blob legacy `routes/quiz.py` used to
+ stuff into the prompt template.
+ - `quiz_attempts` (one row per attempt, filtered to completed
+ attempts): the last N completed attempts, newest first, with
+ accuracy precomputed so the agent doesn't have to.
+
+ Wraps the sync Supabase reads in `asyncio.to_thread` so we don't
+ block the event loop. Failures degrade silently — the agent can
+ still generate a quiz without history (just less adaptive).
+ """
+
+ def _fetch_summary() -> Any:
+ try:
+ rows = table("quiz_context").select(
+ "context_json",
+ filters={
+ "user_id": f"eq.{user_id}",
+ "concept_node_id": f"eq.{concept_node_id}",
+ },
+ limit=1,
+ )
+ return rows[0]["context_json"] if rows else None
+ except Exception:
+ logger.exception(
+ "read_recent_quiz_attempts: quiz_context fetch failed "
+ "user=%s concept=%s",
+ user_id,
+ concept_node_id,
+ )
+ return None
+
+ def _fetch_attempts() -> list[dict[str, Any]]:
+ try:
+ return (
+ table("quiz_attempts").select(
+ "score,total,difficulty,completed_at",
+ filters={
+ "user_id": f"eq.{user_id}",
+ "concept_node_id": f"eq.{concept_node_id}",
+ # Only count completed attempts. PostgREST `not.is.null`
+ # filters out rows where completed_at is NULL, which is
+ # how `routes/quiz.py:generate_quiz` marks an in-flight
+ # attempt before submission.
+ "completed_at": "not.is.null",
+ },
+ order="completed_at.desc",
+ limit=_RECENT_ATTEMPTS_LIMIT,
+ )
+ or []
+ )
+ except Exception:
+ logger.exception(
+ "read_recent_quiz_attempts: quiz_attempts fetch failed "
+ "user=%s concept=%s",
+ user_id,
+ concept_node_id,
+ )
+ return []
+
+ summary_raw, attempt_rows = await asyncio.gather(
+ asyncio.to_thread(_fetch_summary),
+ asyncio.to_thread(_fetch_attempts),
+ )
+
+ attempts: list[RecentQuizAttempt] = []
+ for r in attempt_rows:
+ raw_score = r.get("score")
+ raw_total = r.get("total")
+ if raw_score is None or raw_total is None:
+ # `submit_quiz` writes score+total atomically, so a row with
+ # `completed_at IS NOT NULL` but a null score/total is
+ # corruption (or an out-of-band edit). Drop it rather than
+ # coercing to 0/0 — feeding the LLM a bogus 0% accuracy
+ # could trigger a spurious adaptive downshift.
+ logger.warning(
+ "read_recent_quiz_attempts: dropping row with null "
+ "score/total (score=%r, total=%r) user=%s concept=%s",
+ raw_score,
+ raw_total,
+ user_id,
+ concept_node_id,
+ )
+ continue
+ try:
+ score = int(raw_score)
+ total = int(raw_total)
+ except (TypeError, ValueError):
+ continue
+ if total <= 0:
+ # Skip rows that look incomplete — accuracy is undefined and
+ # the agent shouldn't have to guess.
+ continue
+ if score < 0 or score > total:
+ # Corrupt row (score outside [0, total]). Drop entirely
+ # rather than passing impossible numbers to the LLM —
+ # `score=7, total=5` would prompt the agent to wonder
+ # whether to trust the data at all.
+ logger.warning(
+ "read_recent_quiz_attempts: dropping corrupt row "
+ "(score=%d outside [0, total=%d]) user=%s concept=%s",
+ score,
+ total,
+ user_id,
+ concept_node_id,
+ )
+ continue
+ accuracy = score / total
+ attempts.append(
+ RecentQuizAttempt(
+ score=score,
+ total=total,
+ difficulty=str(r.get("difficulty") or ""),
+ completed_at=r.get("completed_at"),
+ accuracy=round(accuracy, 4),
+ )
+ )
+
+ return QuizHistory(
+ summary=_coerce_summary(summary_raw),
+ recent_attempts=attempts,
+ )
+
+
+async def read_recent_quiz_attempts_tool(
+ ctx: RunContext[SaplingDeps],
+ concept_node_id: str,
+) -> QuizHistory:
+ """Returns this student's history on one concept: a `summary`
+ string digesting their prior mistakes (mine for distractor
+ inspiration) and `recent_attempts` — the last 5 completed quiz
+ attempts on this concept, newest first, with `accuracy` precomputed
+ so you can apply the adaptive-difficulty rule directly. Empty
+ history on first attempt. Pass the `concept_node_id` from the
+ user message; user identity is taken from context.
+ """
+ # user_id comes from ctx.deps so a tool call can't cross users.
+ return await read_recent_quiz_attempts(ctx.deps.user_id, concept_node_id)

backend/routes/quiz.py

@@ -166,11 +166,15 @@ async def _quiz_via_agent(
  supabase=None,
  request_id=request_id,
  )
+ # Keep this message routing-only; the workflow + adaptive rules
+ # live in the system prompt. We just hand the agent the inputs it
+ # needs and trust the prompt to drive tool calls.
  user_message = (
  f"Generate {num_questions} {difficulty} questions for the student. "
- f"The target concept is '{concept_name}' (concept_node_id={concept_node_id}). "
- f"Call read_concepts_for_user to find the student's weakest concepts in this course "
- f"and bias the question mix toward those."
+ f"The target concept is '{concept_name}' "
+ f"(concept_node_id={concept_node_id}). Follow the workflow in your "
+ f"system prompt; pass concept_node_id='{concept_node_id}' to "
+ f"read_recent_quiz_attempts."
  )
  if use_shared_context:
  user_message += (

backend/agents/quiz.py

@@ -24,6 +24,7 @@
  read_concepts_for_user_tool,
  read_misconceptions_for_course_tool,
 )
+from agents.tools.quiz_history import read_recent_quiz_attempts_tool
 
 
 # Difficulty + question type are Literals so Gemini's enum constraint
@@ -69,31 +70,63 @@ class Quiz(BaseModel):
 _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"
+ "mastery on, OR address a class-level misconception you've seen, "
+ "OR revive a concept the student hasn't reviewed in a while.\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"
+ " weakest first). Each concept also carries `last_reviewed_at`, "
+ " which you use for spaced repetition (see rules below).\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"
+ "3. Call `read_recent_quiz_attempts(concept_node_id)` for the "
+ " target concept_node_id given in the user message. The "
+ " `summary` is a digest of past mistakes the student has made "
+ " on this concept — mine it for distractor inspiration. The "
+ " `recent_attempts` list (newest first) drives adaptive "
+ " difficulty (see rules below).\n"
+ "4. Compose `Quiz.questions` so the WEAKEST and STALEST concepts "
+ " get the most questions, AND each item's `concept` field "
+ " exactly matches a concept_name returned by tool 1.\n\n"
+ "Concept-selection rules (combine all three signals):\n"
+ "- Bias question count toward the lowest-mastery concepts (the "
+ " weakest first in the tool 1 return).\n"
+ "- SPACED REPETITION: also surface concepts whose "
+ " `last_reviewed_at` is older than ~7 days, even if their "
+ " mastery is mid-tier — they're due for review and decay over "
+ " time. Concepts with `last_reviewed_at = null` are unreviewed; "
+ " treat them as stale.\n"
+ "- Don't drop high-mastery, recently-reviewed concepts entirely; "
+ " include 1 question on a strong-and-fresh concept to keep the "
+ " quiz from feeling punishing.\n\n"
+ "Adaptive-difficulty rules (use `recent_attempts.accuracy`):\n"
+ "- If the most recent 2-3 attempts on this concept averaged < "
+ " 0.5 accuracy, drop the difficulty mix one step from what the "
+ " user asked (hard -> medium, medium -> easy, easy stays easy). "
+ " The student is struggling; keep them on track.\n"
+ "- If the most recent 3 attempts all scored >= 0.8, you may "
+ " include 1-2 questions one step harder than the requested "
+ " difficulty to push them.\n"
+ "- If `recent_attempts` is empty (first attempt), honor the "
+ " user-requested difficulty exactly.\n"
+ "- Never override the user-requested difficulty by more than one "
+ " step in either direction. Stay close to what they asked for.\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"
+ " noise. Combine signals from `read_misconceptions_for_course` "
+ " (class-wide) and `read_recent_quiz_attempts.summary` "
+ " (this student's prior errors) when writing them.\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."
+ "- difficulty: align with the student's mastery on the concept "
+ " AND the adaptive-difficulty rules above.\n\n"
+ "Honor the requested num_questions. Don't invent concepts the "
+ "student doesn't have."
 )
 _PROMPT_HASH = hashlib.sha256(_SYSTEM_PROMPT.encode("utf-8")).hexdigest()[:12]
 
@@ -107,5 +140,6 @@ class Quiz(BaseModel):
  tools=[
  read_concepts_for_user_tool,
  read_misconceptions_for_course_tool,
+ read_recent_quiz_attempts_tool,
  ],
 )

backend/agents/tools/quiz_history.py

@@ -0,0 +1,234 @@
+"""Quiz-history read tool for the quiz agent.
+
+Surfaces what the student previously got wrong on a concept and how
+their last few attempts scored. The agent uses this for two things:
+
+1. Targeting — write distractors that mirror the student's prior
+ mistakes (the LLM-generated `summary` from `quiz_context`
+ captures patterns rolled up across past attempts).
+2. Adaptive difficulty — read the last few `quiz_attempts` rows and
+ step difficulty down when the student has been struggling, up
+ when they've been crushing it.
+
+The pure async function is callable from routes/tests; the *_tool
+wrapper registers 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__)
+
+
+# How many past attempts the agent gets to see. 5 is enough to spot a
+# trend without flooding the prompt with state. Older attempts are
+# already rolled into `summary` by the post-quiz context update job.
+_RECENT_ATTEMPTS_LIMIT = 5
+
+
+class RecentQuizAttempt(BaseModel):
+ """One past attempt's headline numbers."""
+
+ score: int = Field(ge=0)
+ total: int = Field(ge=0)
+ difficulty: str
+ completed_at: str | None = None
+ accuracy: float = Field(ge=0.0, le=1.0)
+
+
+class QuizHistory(BaseModel):
+ """The agent's view of a student's history on one concept."""
+
+ # LLM-generated digest of past quiz mistakes/patterns for this
+ # (user, concept). Populated by the background context-update job
+ # in routes/quiz.py:submit_quiz. May be None on a first attempt.
+ summary: str | None = None
+ # Most recent attempts, newest first. Empty on first attempt.
+ recent_attempts: list[RecentQuizAttempt] = Field(default_factory=list)
+
+
+def _coerce_summary(ctx: Any) -> str | None:
+ """quiz_context.context_json is free-form (whatever the post-submit
+ LLM produced). Different prompt versions have stored either a flat
+ string or a small dict. Coerce to a single string the agent can
+ reason over, or None if there's nothing useful."""
+ if not ctx:
+ return None
+ if isinstance(ctx, str):
+ text = ctx.strip()
+ return text or None
+ if isinstance(ctx, dict):
+ # Common shapes: {"summary": "..."}, {"notes": "..."},
+ # {"misconceptions": [...], "weak_areas": [...]}.
+ for key in ("summary", "notes", "context", "digest"):
+ v = ctx.get(key)
+ if isinstance(v, str) and v.strip():
+ return v.strip()
+ # Fall back to flattening list-of-strings entries so the agent
+ # at least sees the misconceptions/weak_areas the prior job
+ # extracted, even when no top-level summary string exists.
+ parts: list[str] = []
+ for key in ("misconceptions", "weak_areas", "common_errors"):
+ for item in ctx.get(key) or []:
+ if isinstance(item, str) and item.strip():
+ parts.append(f"- {item.strip()}")
+ return "\n".join(parts) or None
+ return None
+
+
+async def read_recent_quiz_attempts(
+ user_id: str,
+ concept_node_id: str,
+) -> QuizHistory:
+ """Return the agent's view of a student's history on one concept.
+
+ Reads two sources:
+
+ - `quiz_context` (one row per (user, concept)): the rolling
+ LLM-generated digest of what the student has been getting
+ wrong. This is the same blob legacy `routes/quiz.py` used to
+ stuff into the prompt template.
+ - `quiz_attempts` (one row per attempt, filtered to completed
+ attempts): the last N completed attempts, newest first, with
+ accuracy precomputed so the agent doesn't have to.
+
+ Wraps the sync Supabase reads in `asyncio.to_thread` so we don't
+ block the event loop. Failures degrade silently — the agent can
+ still generate a quiz without history (just less adaptive).
+ """
+
+ def _fetch_summary() -> Any:
+ try:
+ rows = table("quiz_context").select(
+ "context_json",
+ filters={
+ "user_id": f"eq.{user_id}",
+ "concept_node_id": f"eq.{concept_node_id}",
+ },
+ limit=1,
+ )
+ return rows[0]["context_json"] if rows else None
+ except Exception:
+ logger.exception(
+ "read_recent_quiz_attempts: quiz_context fetch failed "
+ "user=%s concept=%s",
+ user_id,
+ concept_node_id,
+ )
+ return None
+
+ def _fetch_attempts() -> list[dict[str, Any]]:
+ try:
+ return (
+ table("quiz_attempts").select(
+ "score,total,difficulty,completed_at",
+ filters={
+ "user_id": f"eq.{user_id}",
+ "concept_node_id": f"eq.{concept_node_id}",
+ # Only count completed attempts. PostgREST `not.is.null`
+ # filters out rows where completed_at is NULL, which is
+ # how `routes/quiz.py:generate_quiz` marks an in-flight
+ # attempt before submission.
+ "completed_at": "not.is.null",
+ },
+ order="completed_at.desc",
+ limit=_RECENT_ATTEMPTS_LIMIT,
+ )
+ or []
+ )
+ except Exception:
+ logger.exception(
+ "read_recent_quiz_attempts: quiz_attempts fetch failed "
+ "user=%s concept=%s",
+ user_id,
+ concept_node_id,
+ )
+ return []
+
+ summary_raw, attempt_rows = await asyncio.gather(
+ asyncio.to_thread(_fetch_summary),
+ asyncio.to_thread(_fetch_attempts),
+ )
+
+ attempts: list[RecentQuizAttempt] = []
+ for r in attempt_rows:
+ raw_score = r.get("score")
+ raw_total = r.get("total")
+ if raw_score is None or raw_total is None:
+ # `submit_quiz` writes score+total atomically, so a row with
+ # `completed_at IS NOT NULL` but a null score/total is
+ # corruption (or an out-of-band edit). Drop it rather than
+ # coercing to 0/0 — feeding the LLM a bogus 0% accuracy
+ # could trigger a spurious adaptive downshift.
+ logger.warning(
+ "read_recent_quiz_attempts: dropping row with null "
+ "score/total (score=%r, total=%r) user=%s concept=%s",
+ raw_score,
+ raw_total,
+ user_id,
+ concept_node_id,
+ )
+ continue
+ try:
+ score = int(raw_score)
+ total = int(raw_total)
+ except (TypeError, ValueError):
+ continue
+ if total <= 0:
+ # Skip rows that look incomplete — accuracy is undefined and
+ # the agent shouldn't have to guess.
+ continue
+ if score < 0 or score > total:
+ # Corrupt row (score outside [0, total]). Drop entirely
+ # rather than passing impossible numbers to the LLM —
+ # `score=7, total=5` would prompt the agent to wonder
+ # whether to trust the data at all.
+ logger.warning(
+ "read_recent_quiz_attempts: dropping corrupt row "
+ "(score=%d outside [0, total=%d]) user=%s concept=%s",
+ score,
+ total,
+ user_id,
+ concept_node_id,
+ )
+ continue
+ accuracy = score / total
+ attempts.append(
+ RecentQuizAttempt(
+ score=score,
+ total=total,
+ difficulty=str(r.get("difficulty") or ""),
+ completed_at=r.get("completed_at"),
+ accuracy=round(accuracy, 4),
+ )
+ )
+
+ return QuizHistory(
+ summary=_coerce_summary(summary_raw),
+ recent_attempts=attempts,
+ )
+
+
+async def read_recent_quiz_attempts_tool(
+ ctx: RunContext[SaplingDeps],
+ concept_node_id: str,
+) -> QuizHistory:
+ """Returns this student's history on one concept: a `summary`
+ string digesting their prior mistakes (mine for distractor
+ inspiration) and `recent_attempts` — the last 5 completed quiz
+ attempts on this concept, newest first, with `accuracy` precomputed
+ so you can apply the adaptive-difficulty rule directly. Empty
+ history on first attempt. Pass the `concept_node_id` from the
+ user message; user identity is taken from context.
+ """
+ # user_id comes from ctx.deps so a tool call can't cross users.
+ return await read_recent_quiz_attempts(ctx.deps.user_id, concept_node_id)

backend/routes/quiz.py

@@ -166,11 +166,15 @@ async def _quiz_via_agent(
  supabase=None,
  request_id=request_id,
  )
+ # Keep this message routing-only; the workflow + adaptive rules
+ # live in the system prompt. We just hand the agent the inputs it
+ # needs and trust the prompt to drive tool calls.
  user_message = (
  f"Generate {num_questions} {difficulty} questions for the student. "
- f"The target concept is '{concept_name}' (concept_node_id={concept_node_id}). "
- f"Call read_concepts_for_user to find the student's weakest concepts in this course "
- f"and bias the question mix toward those."
+ f"The target concept is '{concept_name}' "
+ f"(concept_node_id={concept_node_id}). Follow the workflow in your "
+ f"system prompt; pass concept_node_id='{concept_node_id}' to "
+ f"read_recent_quiz_attempts."
  )
  if use_shared_context:
  user_message += (

backend/agents/quiz.py

@@ -24,6 +24,7 @@
  read_concepts_for_user_tool,
  read_misconceptions_for_course_tool,
 )
+from agents.tools.quiz_history import read_recent_quiz_attempts_tool
 
 
 # Difficulty + question type are Literals so Gemini's enum constraint
@@ -69,31 +70,63 @@ class Quiz(BaseModel):
 _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"
+ "mastery on, OR address a class-level misconception you've seen, "
+ "OR revive a concept the student hasn't reviewed in a while.\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"
+ " weakest first). Each concept also carries `last_reviewed_at`, "
+ " which you use for spaced repetition (see rules below).\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"
+ "3. Call `read_recent_quiz_attempts(concept_node_id)` for the "
+ " target concept_node_id given in the user message. The "
+ " `summary` is a digest of past mistakes the student has made "
+ " on this concept — mine it for distractor inspiration. The "
+ " `recent_attempts` list (newest first) drives adaptive "
+ " difficulty (see rules below).\n"
+ "4. Compose `Quiz.questions` so the WEAKEST and STALEST concepts "
+ " get the most questions, AND each item's `concept` field "
+ " exactly matches a concept_name returned by tool 1.\n\n"
+ "Concept-selection rules (combine all three signals):\n"
+ "- Bias question count toward the lowest-mastery concepts (the "
+ " weakest first in the tool 1 return).\n"
+ "- SPACED REPETITION: also surface concepts whose "
+ " `last_reviewed_at` is older than ~7 days, even if their "
+ " mastery is mid-tier — they're due for review and decay over "
+ " time. Concepts with `last_reviewed_at = null` are unreviewed; "
+ " treat them as stale.\n"
+ "- Don't drop high-mastery, recently-reviewed concepts entirely; "
+ " include 1 question on a strong-and-fresh concept to keep the "
+ " quiz from feeling punishing.\n\n"
+ "Adaptive-difficulty rules (use `recent_attempts.accuracy`):\n"
+ "- If the most recent 2-3 attempts on this concept averaged < "
+ " 0.5 accuracy, drop the difficulty mix one step from what the "
+ " user asked (hard -> medium, medium -> easy, easy stays easy). "
+ " The student is struggling; keep them on track.\n"
+ "- If the most recent 3 attempts all scored >= 0.8, you may "
+ " include 1-2 questions one step harder than the requested "
+ " difficulty to push them.\n"
+ "- If `recent_attempts` is empty (first attempt), honor the "
+ " user-requested difficulty exactly.\n"
+ "- Never override the user-requested difficulty by more than one "
+ " step in either direction. Stay close to what they asked for.\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"
+ " noise. Combine signals from `read_misconceptions_for_course` "
+ " (class-wide) and `read_recent_quiz_attempts.summary` "
+ " (this student's prior errors) when writing them.\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."
+ "- difficulty: align with the student's mastery on the concept "
+ " AND the adaptive-difficulty rules above.\n\n"
+ "Honor the requested num_questions. Don't invent concepts the "
+ "student doesn't have."
 )
 _PROMPT_HASH = hashlib.sha256(_SYSTEM_PROMPT.encode("utf-8")).hexdigest()[:12]
 
@@ -107,5 +140,6 @@ class Quiz(BaseModel):
  tools=[
  read_concepts_for_user_tool,
  read_misconceptions_for_course_tool,
+ read_recent_quiz_attempts_tool,
  ],
 )

backend/agents/tools/quiz_history.py

@@ -0,0 +1,234 @@
+"""Quiz-history read tool for the quiz agent.
+
+Surfaces what the student previously got wrong on a concept and how
+their last few attempts scored. The agent uses this for two things:
+
+1. Targeting — write distractors that mirror the student's prior
+ mistakes (the LLM-generated `summary` from `quiz_context`
+ captures patterns rolled up across past attempts).
+2. Adaptive difficulty — read the last few `quiz_attempts` rows and
+ step difficulty down when the student has been struggling, up
+ when they've been crushing it.
+
+The pure async function is callable from routes/tests; the *_tool
+wrapper registers 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__)
+
+
+# How many past attempts the agent gets to see. 5 is enough to spot a
+# trend without flooding the prompt with state. Older attempts are
+# already rolled into `summary` by the post-quiz context update job.
+_RECENT_ATTEMPTS_LIMIT = 5
+
+
+class RecentQuizAttempt(BaseModel):
+ """One past attempt's headline numbers."""
+
+ score: int = Field(ge=0)
+ total: int = Field(ge=0)
+ difficulty: str
+ completed_at: str | None = None
+ accuracy: float = Field(ge=0.0, le=1.0)
+
+
+class QuizHistory(BaseModel):
+ """The agent's view of a student's history on one concept."""
+
+ # LLM-generated digest of past quiz mistakes/patterns for this
+ # (user, concept). Populated by the background context-update job
+ # in routes/quiz.py:submit_quiz. May be None on a first attempt.
+ summary: str | None = None
+ # Most recent attempts, newest first. Empty on first attempt.
+ recent_attempts: list[RecentQuizAttempt] = Field(default_factory=list)
+
+
+def _coerce_summary(ctx: Any) -> str | None:
+ """quiz_context.context_json is free-form (whatever the post-submit
+ LLM produced). Different prompt versions have stored either a flat
+ string or a small dict. Coerce to a single string the agent can
+ reason over, or None if there's nothing useful."""
+ if not ctx:
+ return None
+ if isinstance(ctx, str):
+ text = ctx.strip()
+ return text or None
+ if isinstance(ctx, dict):
+ # Common shapes: {"summary": "..."}, {"notes": "..."},
+ # {"misconceptions": [...], "weak_areas": [...]}.
+ for key in ("summary", "notes", "context", "digest"):
+ v = ctx.get(key)
+ if isinstance(v, str) and v.strip():
+ return v.strip()
+ # Fall back to flattening list-of-strings entries so the agent
+ # at least sees the misconceptions/weak_areas the prior job
+ # extracted, even when no top-level summary string exists.
+ parts: list[str] = []
+ for key in ("misconceptions", "weak_areas", "common_errors"):
+ for item in ctx.get(key) or []:
+ if isinstance(item, str) and item.strip():
+ parts.append(f"- {item.strip()}")
+ return "\n".join(parts) or None
+ return None
+
+
+async def read_recent_quiz_attempts(
+ user_id: str,
+ concept_node_id: str,
+) -> QuizHistory:
+ """Return the agent's view of a student's history on one concept.
+
+ Reads two sources:
+
+ - `quiz_context` (one row per (user, concept)): the rolling
+ LLM-generated digest of what the student has been getting
+ wrong. This is the same blob legacy `routes/quiz.py` used to
+ stuff into the prompt template.
+ - `quiz_attempts` (one row per attempt, filtered to completed
+ attempts): the last N completed attempts, newest first, with
+ accuracy precomputed so the agent doesn't have to.
+
+ Wraps the sync Supabase reads in `asyncio.to_thread` so we don't
+ block the event loop. Failures degrade silently — the agent can
+ still generate a quiz without history (just less adaptive).
+ """
+
+ def _fetch_summary() -> Any:
+ try:
+ rows = table("quiz_context").select(
+ "context_json",
+ filters={
+ "user_id": f"eq.{user_id}",
+ "concept_node_id": f"eq.{concept_node_id}",
+ },
+ limit=1,
+ )
+ return rows[0]["context_json"] if rows else None
+ except Exception:
+ logger.exception(
+ "read_recent_quiz_attempts: quiz_context fetch failed "
+ "user=%s concept=%s",
+ user_id,
+ concept_node_id,
+ )
+ return None
+
+ def _fetch_attempts() -> list[dict[str, Any]]:
+ try:
+ return (
+ table("quiz_attempts").select(
+ "score,total,difficulty,completed_at",
+ filters={
+ "user_id": f"eq.{user_id}",
+ "concept_node_id": f"eq.{concept_node_id}",
+ # Only count completed attempts. PostgREST `not.is.null`
+ # filters out rows where completed_at is NULL, which is
+ # how `routes/quiz.py:generate_quiz` marks an in-flight
+ # attempt before submission.
+ "completed_at": "not.is.null",
+ },
+ order="completed_at.desc",
+ limit=_RECENT_ATTEMPTS_LIMIT,
+ )
+ or []
+ )
+ except Exception:
+ logger.exception(
+ "read_recent_quiz_attempts: quiz_attempts fetch failed "
+ "user=%s concept=%s",
+ user_id,
+ concept_node_id,
+ )
+ return []
+
+ summary_raw, attempt_rows = await asyncio.gather(
+ asyncio.to_thread(_fetch_summary),
+ asyncio.to_thread(_fetch_attempts),
+ )
+
+ attempts: list[RecentQuizAttempt] = []
+ for r in attempt_rows:
+ raw_score = r.get("score")
+ raw_total = r.get("total")
+ if raw_score is None or raw_total is None:
+ # `submit_quiz` writes score+total atomically, so a row with
+ # `completed_at IS NOT NULL` but a null score/total is
+ # corruption (or an out-of-band edit). Drop it rather than
+ # coercing to 0/0 — feeding the LLM a bogus 0% accuracy
+ # could trigger a spurious adaptive downshift.
+ logger.warning(
+ "read_recent_quiz_attempts: dropping row with null "
+ "score/total (score=%r, total=%r) user=%s concept=%s",
+ raw_score,
+ raw_total,
+ user_id,
+ concept_node_id,
+ )
+ continue
+ try:
+ score = int(raw_score)
+ total = int(raw_total)
+ except (TypeError, ValueError):
+ continue
+ if total <= 0:
+ # Skip rows that look incomplete — accuracy is undefined and
+ # the agent shouldn't have to guess.
+ continue
+ if score < 0 or score > total:
+ # Corrupt row (score outside [0, total]). Drop entirely
+ # rather than passing impossible numbers to the LLM —
+ # `score=7, total=5` would prompt the agent to wonder
+ # whether to trust the data at all.
+ logger.warning(
+ "read_recent_quiz_attempts: dropping corrupt row "
+ "(score=%d outside [0, total=%d]) user=%s concept=%s",
+ score,
+ total,
+ user_id,
+ concept_node_id,
+ )
+ continue
+ accuracy = score / total
+ attempts.append(
+ RecentQuizAttempt(
+ score=score,
+ total=total,
+ difficulty=str(r.get("difficulty") or ""),
+ completed_at=r.get("completed_at"),
+ accuracy=round(accuracy, 4),
+ )
+ )
+
+ return QuizHistory(
+ summary=_coerce_summary(summary_raw),
+ recent_attempts=attempts,
+ )
+
+
+async def read_recent_quiz_attempts_tool(
+ ctx: RunContext[SaplingDeps],
+ concept_node_id: str,
+) -> QuizHistory:
+ """Returns this student's history on one concept: a `summary`
+ string digesting their prior mistakes (mine for distractor
+ inspiration) and `recent_attempts` — the last 5 completed quiz
+ attempts on this concept, newest first, with `accuracy` precomputed
+ so you can apply the adaptive-difficulty rule directly. Empty
+ history on first attempt. Pass the `concept_node_id` from the
+ user message; user identity is taken from context.
+ """
+ # user_id comes from ctx.deps so a tool call can't cross users.
+ return await read_recent_quiz_attempts(ctx.deps.user_id, concept_node_id)

backend/routes/quiz.py

@@ -166,11 +166,15 @@ async def _quiz_via_agent(
  supabase=None,
  request_id=request_id,
  )
+ # Keep this message routing-only; the workflow + adaptive rules
+ # live in the system prompt. We just hand the agent the inputs it
+ # needs and trust the prompt to drive tool calls.
  user_message = (
  f"Generate {num_questions} {difficulty} questions for the student. "
- f"The target concept is '{concept_name}' (concept_node_id={concept_node_id}). "
- f"Call read_concepts_for_user to find the student's weakest concepts in this course "
- f"and bias the question mix toward those."
+ f"The target concept is '{concept_name}' "
+ f"(concept_node_id={concept_node_id}). Follow the workflow in your "
+ f"system prompt; pass concept_node_id='{concept_node_id}' to "
+ f"read_recent_quiz_attempts."
  )
  if use_shared_context:
  user_message += (

backend/agents/quiz.py

@@ -24,6 +24,7 @@
  read_concepts_for_user_tool,
  read_misconceptions_for_course_tool,
 )
+from agents.tools.quiz_history import read_recent_quiz_attempts_tool
 
 
 # Difficulty + question type are Literals so Gemini's enum constraint
@@ -69,31 +70,63 @@ class Quiz(BaseModel):
 _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"
+ "mastery on, OR address a class-level misconception you've seen, "
+ "OR revive a concept the student hasn't reviewed in a while.\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"
+ " weakest first). Each concept also carries `last_reviewed_at`, "
+ " which you use for spaced repetition (see rules below).\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"
+ "3. Call `read_recent_quiz_attempts(concept_node_id)` for the "
+ " target concept_node_id given in the user message. The "
+ " `summary` is a digest of past mistakes the student has made "
+ " on this concept — mine it for distractor inspiration. The "
+ " `recent_attempts` list (newest first) drives adaptive "
+ " difficulty (see rules below).\n"
+ "4. Compose `Quiz.questions` so the WEAKEST and STALEST concepts "
+ " get the most questions, AND each item's `concept` field "
+ " exactly matches a concept_name returned by tool 1.\n\n"
+ "Concept-selection rules (combine all three signals):\n"
+ "- Bias question count toward the lowest-mastery concepts (the "
+ " weakest first in the tool 1 return).\n"
+ "- SPACED REPETITION: also surface concepts whose "
+ " `last_reviewed_at` is older than ~7 days, even if their "
+ " mastery is mid-tier — they're due for review and decay over "
+ " time. Concepts with `last_reviewed_at = null` are unreviewed; "
+ " treat them as stale.\n"
+ "- Don't drop high-mastery, recently-reviewed concepts entirely; "
+ " include 1 question on a strong-and-fresh concept to keep the "
+ " quiz from feeling punishing.\n\n"
+ "Adaptive-difficulty rules (use `recent_attempts.accuracy`):\n"
+ "- If the most recent 2-3 attempts on this concept averaged < "
+ " 0.5 accuracy, drop the difficulty mix one step from what the "
+ " user asked (hard -> medium, medium -> easy, easy stays easy). "
+ " The student is struggling; keep them on track.\n"
+ "- If the most recent 3 attempts all scored >= 0.8, you may "
+ " include 1-2 questions one step harder than the requested "
+ " difficulty to push them.\n"
+ "- If `recent_attempts` is empty (first attempt), honor the "
+ " user-requested difficulty exactly.\n"
+ "- Never override the user-requested difficulty by more than one "
+ " step in either direction. Stay close to what they asked for.\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"
+ " noise. Combine signals from `read_misconceptions_for_course` "
+ " (class-wide) and `read_recent_quiz_attempts.summary` "
+ " (this student's prior errors) when writing them.\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."
+ "- difficulty: align with the student's mastery on the concept "
+ " AND the adaptive-difficulty rules above.\n\n"
+ "Honor the requested num_questions. Don't invent concepts the "
+ "student doesn't have."
 )
 _PROMPT_HASH = hashlib.sha256(_SYSTEM_PROMPT.encode("utf-8")).hexdigest()[:12]
 
@@ -107,5 +140,6 @@ class Quiz(BaseModel):
  tools=[
  read_concepts_for_user_tool,
  read_misconceptions_for_course_tool,
+ read_recent_quiz_attempts_tool,
  ],
 )

backend/agents/tools/quiz_history.py

@@ -0,0 +1,234 @@
+"""Quiz-history read tool for the quiz agent.
+
+Surfaces what the student previously got wrong on a concept and how
+their last few attempts scored. The agent uses this for two things:
+
+1. Targeting — write distractors that mirror the student's prior
+ mistakes (the LLM-generated `summary` from `quiz_context`
+ captures patterns rolled up across past attempts).
+2. Adaptive difficulty — read the last few `quiz_attempts` rows and
+ step difficulty down when the student has been struggling, up
+ when they've been crushing it.
+
+The pure async function is callable from routes/tests; the *_tool
+wrapper registers 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__)
+
+
+# How many past attempts the agent gets to see. 5 is enough to spot a
+# trend without flooding the prompt with state. Older attempts are
+# already rolled into `summary` by the post-quiz context update job.
+_RECENT_ATTEMPTS_LIMIT = 5
+
+
+class RecentQuizAttempt(BaseModel):
+ """One past attempt's headline numbers."""
+
+ score: int = Field(ge=0)
+ total: int = Field(ge=0)
+ difficulty: str
+ completed_at: str | None = None
+ accuracy: float = Field(ge=0.0, le=1.0)
+
+
+class QuizHistory(BaseModel):
+ """The agent's view of a student's history on one concept."""
+
+ # LLM-generated digest of past quiz mistakes/patterns for this
+ # (user, concept). Populated by the background context-update job
+ # in routes/quiz.py:submit_quiz. May be None on a first attempt.
+ summary: str | None = None
+ # Most recent attempts, newest first. Empty on first attempt.
+ recent_attempts: list[RecentQuizAttempt] = Field(default_factory=list)
+
+
+def _coerce_summary(ctx: Any) -> str | None:
+ """quiz_context.context_json is free-form (whatever the post-submit
+ LLM produced). Different prompt versions have stored either a flat
+ string or a small dict. Coerce to a single string the agent can
+ reason over, or None if there's nothing useful."""
+ if not ctx:
+ return None
+ if isinstance(ctx, str):
+ text = ctx.strip()
+ return text or None
+ if isinstance(ctx, dict):
+ # Common shapes: {"summary": "..."}, {"notes": "..."},
+ # {"misconceptions": [...], "weak_areas": [...]}.
+ for key in ("summary", "notes", "context", "digest"):
+ v = ctx.get(key)
+ if isinstance(v, str) and v.strip():
+ return v.strip()
+ # Fall back to flattening list-of-strings entries so the agent
+ # at least sees the misconceptions/weak_areas the prior job
+ # extracted, even when no top-level summary string exists.
+ parts: list[str] = []
+ for key in ("misconceptions", "weak_areas", "common_errors"):
+ for item in ctx.get(key) or []:
+ if isinstance(item, str) and item.strip():
+ parts.append(f"- {item.strip()}")
+ return "\n".join(parts) or None
+ return None
+
+
+async def read_recent_quiz_attempts(
+ user_id: str,
+ concept_node_id: str,
+) -> QuizHistory:
+ """Return the agent's view of a student's history on one concept.
+
+ Reads two sources:
+
+ - `quiz_context` (one row per (user, concept)): the rolling
+ LLM-generated digest of what the student has been getting
+ wrong. This is the same blob legacy `routes/quiz.py` used to
+ stuff into the prompt template.
+ - `quiz_attempts` (one row per attempt, filtered to completed
+ attempts): the last N completed attempts, newest first, with
+ accuracy precomputed so the agent doesn't have to.
+
+ Wraps the sync Supabase reads in `asyncio.to_thread` so we don't
+ block the event loop. Failures degrade silently — the agent can
+ still generate a quiz without history (just less adaptive).
+ """
+
+ def _fetch_summary() -> Any:
+ try:
+ rows = table("quiz_context").select(
+ "context_json",
+ filters={
+ "user_id": f"eq.{user_id}",
+ "concept_node_id": f"eq.{concept_node_id}",
+ },
+ limit=1,
+ )
+ return rows[0]["context_json"] if rows else None
+ except Exception:
+ logger.exception(
+ "read_recent_quiz_attempts: quiz_context fetch failed "
+ "user=%s concept=%s",
+ user_id,
+ concept_node_id,
+ )
+ return None
+
+ def _fetch_attempts() -> list[dict[str, Any]]:
+ try:
+ return (
+ table("quiz_attempts").select(
+ "score,total,difficulty,completed_at",
+ filters={
+ "user_id": f"eq.{user_id}",
+ "concept_node_id": f"eq.{concept_node_id}",
+ # Only count completed attempts. PostgREST `not.is.null`
+ # filters out rows where completed_at is NULL, which is
+ # how `routes/quiz.py:generate_quiz` marks an in-flight
+ # attempt before submission.
+ "completed_at": "not.is.null",
+ },
+ order="completed_at.desc",
+ limit=_RECENT_ATTEMPTS_LIMIT,
+ )
+ or []
+ )
+ except Exception:
+ logger.exception(
+ "read_recent_quiz_attempts: quiz_attempts fetch failed "
+ "user=%s concept=%s",
+ user_id,
+ concept_node_id,
+ )
+ return []
+
+ summary_raw, attempt_rows = await asyncio.gather(
+ asyncio.to_thread(_fetch_summary),
+ asyncio.to_thread(_fetch_attempts),
+ )
+
+ attempts: list[RecentQuizAttempt] = []
+ for r in attempt_rows:
+ raw_score = r.get("score")
+ raw_total = r.get("total")
+ if raw_score is None or raw_total is None:
+ # `submit_quiz` writes score+total atomically, so a row with
+ # `completed_at IS NOT NULL` but a null score/total is
+ # corruption (or an out-of-band edit). Drop it rather than
+ # coercing to 0/0 — feeding the LLM a bogus 0% accuracy
+ # could trigger a spurious adaptive downshift.
+ logger.warning(
+ "read_recent_quiz_attempts: dropping row with null "
+ "score/total (score=%r, total=%r) user=%s concept=%s",
+ raw_score,
+ raw_total,
+ user_id,
+ concept_node_id,
+ )
+ continue
+ try:
+ score = int(raw_score)
+ total = int(raw_total)
+ except (TypeError, ValueError):
+ continue
+ if total <= 0:
+ # Skip rows that look incomplete — accuracy is undefined and
+ # the agent shouldn't have to guess.
+ continue
+ if score < 0 or score > total:
+ # Corrupt row (score outside [0, total]). Drop entirely
+ # rather than passing impossible numbers to the LLM —
+ # `score=7, total=5` would prompt the agent to wonder
+ # whether to trust the data at all.
+ logger.warning(
+ "read_recent_quiz_attempts: dropping corrupt row "
+ "(score=%d outside [0, total=%d]) user=%s concept=%s",
+ score,
+ total,
+ user_id,
+ concept_node_id,
+ )
+ continue
+ accuracy = score / total
+ attempts.append(
+ RecentQuizAttempt(
+ score=score,
+ total=total,
+ difficulty=str(r.get("difficulty") or ""),
+ completed_at=r.get("completed_at"),
+ accuracy=round(accuracy, 4),
+ )
+ )
+
+ return QuizHistory(
+ summary=_coerce_summary(summary_raw),
+ recent_attempts=attempts,
+ )
+
+
+async def read_recent_quiz_attempts_tool(
+ ctx: RunContext[SaplingDeps],
+ concept_node_id: str,
+) -> QuizHistory:
+ """Returns this student's history on one concept: a `summary`
+ string digesting their prior mistakes (mine for distractor
+ inspiration) and `recent_attempts` — the last 5 completed quiz
+ attempts on this concept, newest first, with `accuracy` precomputed
+ so you can apply the adaptive-difficulty rule directly. Empty
+ history on first attempt. Pass the `concept_node_id` from the
+ user message; user identity is taken from context.
+ """
+ # user_id comes from ctx.deps so a tool call can't cross users.
+ return await read_recent_quiz_attempts(ctx.deps.user_id, concept_node_id)

backend/routes/quiz.py

@@ -166,11 +166,15 @@ async def _quiz_via_agent(
  supabase=None,
  request_id=request_id,
  )
+ # Keep this message routing-only; the workflow + adaptive rules
+ # live in the system prompt. We just hand the agent the inputs it
+ # needs and trust the prompt to drive tool calls.
  user_message = (
  f"Generate {num_questions} {difficulty} questions for the student. "
- f"The target concept is '{concept_name}' (concept_node_id={concept_node_id}). "
- f"Call read_concepts_for_user to find the student's weakest concepts in this course "
- f"and bias the question mix toward those."
+ f"The target concept is '{concept_name}' "
+ f"(concept_node_id={concept_node_id}). Follow the workflow in your "
+ f"system prompt; pass concept_node_id='{concept_node_id}' to "
+ f"read_recent_quiz_attempts."
  )
  if use_shared_context:
  user_message += (

backend/agents/quiz.py

@@ -24,6 +24,7 @@
  read_concepts_for_user_tool,
  read_misconceptions_for_course_tool,
 )
+from agents.tools.quiz_history import read_recent_quiz_attempts_tool
 
 
 # Difficulty + question type are Literals so Gemini's enum constraint
@@ -69,31 +70,63 @@ class Quiz(BaseModel):
 _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"
+ "mastery on, OR address a class-level misconception you've seen, "
+ "OR revive a concept the student hasn't reviewed in a while.\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"
+ " weakest first). Each concept also carries `last_reviewed_at`, "
+ " which you use for spaced repetition (see rules below).\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"
+ "3. Call `read_recent_quiz_attempts(concept_node_id)` for the "
+ " target concept_node_id given in the user message. The "
+ " `summary` is a digest of past mistakes the student has made "
+ " on this concept — mine it for distractor inspiration. The "
+ " `recent_attempts` list (newest first) drives adaptive "
+ " difficulty (see rules below).\n"
+ "4. Compose `Quiz.questions` so the WEAKEST and STALEST concepts "
+ " get the most questions, AND each item's `concept` field "
+ " exactly matches a concept_name returned by tool 1.\n\n"
+ "Concept-selection rules (combine all three signals):\n"
+ "- Bias question count toward the lowest-mastery concepts (the "
+ " weakest first in the tool 1 return).\n"
+ "- SPACED REPETITION: also surface concepts whose "
+ " `last_reviewed_at` is older than ~7 days, even if their "
+ " mastery is mid-tier — they're due for review and decay over "
+ " time. Concepts with `last_reviewed_at = null` are unreviewed; "
+ " treat them as stale.\n"
+ "- Don't drop high-mastery, recently-reviewed concepts entirely; "
+ " include 1 question on a strong-and-fresh concept to keep the "
+ " quiz from feeling punishing.\n\n"
+ "Adaptive-difficulty rules (use `recent_attempts.accuracy`):\n"
+ "- If the most recent 2-3 attempts on this concept averaged < "
+ " 0.5 accuracy, drop the difficulty mix one step from what the "
+ " user asked (hard -> medium, medium -> easy, easy stays easy). "
+ " The student is struggling; keep them on track.\n"
+ "- If the most recent 3 attempts all scored >= 0.8, you may "
+ " include 1-2 questions one step harder than the requested "
+ " difficulty to push them.\n"
+ "- If `recent_attempts` is empty (first attempt), honor the "
+ " user-requested difficulty exactly.\n"
+ "- Never override the user-requested difficulty by more than one "
+ " step in either direction. Stay close to what they asked for.\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"
+ " noise. Combine signals from `read_misconceptions_for_course` "
+ " (class-wide) and `read_recent_quiz_attempts.summary` "
+ " (this student's prior errors) when writing them.\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."
+ "- difficulty: align with the student's mastery on the concept "
+ " AND the adaptive-difficulty rules above.\n\n"
+ "Honor the requested num_questions. Don't invent concepts the "
+ "student doesn't have."
 )
 _PROMPT_HASH = hashlib.sha256(_SYSTEM_PROMPT.encode("utf-8")).hexdigest()[:12]
 
@@ -107,5 +140,6 @@ class Quiz(BaseModel):
  tools=[
  read_concepts_for_user_tool,
  read_misconceptions_for_course_tool,
+ read_recent_quiz_attempts_tool,
  ],
 )

backend/agents/tools/quiz_history.py

@@ -0,0 +1,234 @@
+"""Quiz-history read tool for the quiz agent.
+
+Surfaces what the student previously got wrong on a concept and how
+their last few attempts scored. The agent uses this for two things:
+
+1. Targeting — write distractors that mirror the student's prior
+ mistakes (the LLM-generated `summary` from `quiz_context`
+ captures patterns rolled up across past attempts).
+2. Adaptive difficulty — read the last few `quiz_attempts` rows and
+ step difficulty down when the student has been struggling, up
+ when they've been crushing it.
+
+The pure async function is callable from routes/tests; the *_tool
+wrapper registers 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__)
+
+
+# How many past attempts the agent gets to see. 5 is enough to spot a
+# trend without flooding the prompt with state. Older attempts are
+# already rolled into `summary` by the post-quiz context update job.
+_RECENT_ATTEMPTS_LIMIT = 5
+
+
+class RecentQuizAttempt(BaseModel):
+ """One past attempt's headline numbers."""
+
+ score: int = Field(ge=0)
+ total: int = Field(ge=0)
+ difficulty: str
+ completed_at: str | None = None
+ accuracy: float = Field(ge=0.0, le=1.0)
+
+
+class QuizHistory(BaseModel):
+ """The agent's view of a student's history on one concept."""
+
+ # LLM-generated digest of past quiz mistakes/patterns for this
+ # (user, concept). Populated by the background context-update job
+ # in routes/quiz.py:submit_quiz. May be None on a first attempt.
+ summary: str | None = None
+ # Most recent attempts, newest first. Empty on first attempt.
+ recent_attempts: list[RecentQuizAttempt] = Field(default_factory=list)
+
+
+def _coerce_summary(ctx: Any) -> str | None:
+ """quiz_context.context_json is free-form (whatever the post-submit
+ LLM produced). Different prompt versions have stored either a flat
+ string or a small dict. Coerce to a single string the agent can
+ reason over, or None if there's nothing useful."""
+ if not ctx:
+ return None
+ if isinstance(ctx, str):
+ text = ctx.strip()
+ return text or None
+ if isinstance(ctx, dict):
+ # Common shapes: {"summary": "..."}, {"notes": "..."},
+ # {"misconceptions": [...], "weak_areas": [...]}.
+ for key in ("summary", "notes", "context", "digest"):
+ v = ctx.get(key)
+ if isinstance(v, str) and v.strip():
+ return v.strip()
+ # Fall back to flattening list-of-strings entries so the agent
+ # at least sees the misconceptions/weak_areas the prior job
+ # extracted, even when no top-level summary string exists.
+ parts: list[str] = []
+ for key in ("misconceptions", "weak_areas", "common_errors"):
+ for item in ctx.get(key) or []:
+ if isinstance(item, str) and item.strip():
+ parts.append(f"- {item.strip()}")
+ return "\n".join(parts) or None
+ return None
+
+
+async def read_recent_quiz_attempts(
+ user_id: str,
+ concept_node_id: str,
+) -> QuizHistory:
+ """Return the agent's view of a student's history on one concept.
+
+ Reads two sources:
+
+ - `quiz_context` (one row per (user, concept)): the rolling
+ LLM-generated digest of what the student has been getting
+ wrong. This is the same blob legacy `routes/quiz.py` used to
+ stuff into the prompt template.
+ - `quiz_attempts` (one row per attempt, filtered to completed
+ attempts): the last N completed attempts, newest first, with
+ accuracy precomputed so the agent doesn't have to.
+
+ Wraps the sync Supabase reads in `asyncio.to_thread` so we don't
+ block the event loop. Failures degrade silently — the agent can
+ still generate a quiz without history (just less adaptive).
+ """
+
+ def _fetch_summary() -> Any:
+ try:
+ rows = table("quiz_context").select(
+ "context_json",
+ filters={
+ "user_id": f"eq.{user_id}",
+ "concept_node_id": f"eq.{concept_node_id}",
+ },
+ limit=1,
+ )
+ return rows[0]["context_json"] if rows else None
+ except Exception:
+ logger.exception(
+ "read_recent_quiz_attempts: quiz_context fetch failed "
+ "user=%s concept=%s",
+ user_id,
+ concept_node_id,
+ )
+ return None
+
+ def _fetch_attempts() -> list[dict[str, Any]]:
+ try:
+ return (
+ table("quiz_attempts").select(
+ "score,total,difficulty,completed_at",
+ filters={
+ "user_id": f"eq.{user_id}",
+ "concept_node_id": f"eq.{concept_node_id}",
+ # Only count completed attempts. PostgREST `not.is.null`
+ # filters out rows where completed_at is NULL, which is
+ # how `routes/quiz.py:generate_quiz` marks an in-flight
+ # attempt before submission.
+ "completed_at": "not.is.null",
+ },
+ order="completed_at.desc",
+ limit=_RECENT_ATTEMPTS_LIMIT,
+ )
+ or []
+ )
+ except Exception:
+ logger.exception(
+ "read_recent_quiz_attempts: quiz_attempts fetch failed "
+ "user=%s concept=%s",
+ user_id,
+ concept_node_id,
+ )
+ return []
+
+ summary_raw, attempt_rows = await asyncio.gather(
+ asyncio.to_thread(_fetch_summary),
+ asyncio.to_thread(_fetch_attempts),
+ )
+
+ attempts: list[RecentQuizAttempt] = []
+ for r in attempt_rows:
+ raw_score = r.get("score")
+ raw_total = r.get("total")
+ if raw_score is None or raw_total is None:
+ # `submit_quiz` writes score+total atomically, so a row with
+ # `completed_at IS NOT NULL` but a null score/total is
+ # corruption (or an out-of-band edit). Drop it rather than
+ # coercing to 0/0 — feeding the LLM a bogus 0% accuracy
+ # could trigger a spurious adaptive downshift.
+ logger.warning(
+ "read_recent_quiz_attempts: dropping row with null "
+ "score/total (score=%r, total=%r) user=%s concept=%s",
+ raw_score,
+ raw_total,
+ user_id,
+ concept_node_id,
+ )
+ continue
+ try:
+ score = int(raw_score)
+ total = int(raw_total)
+ except (TypeError, ValueError):
+ continue
+ if total <= 0:
+ # Skip rows that look incomplete — accuracy is undefined and
+ # the agent shouldn't have to guess.
+ continue
+ if score < 0 or score > total:
+ # Corrupt row (score outside [0, total]). Drop entirely
+ # rather than passing impossible numbers to the LLM —
+ # `score=7, total=5` would prompt the agent to wonder
+ # whether to trust the data at all.
+ logger.warning(
+ "read_recent_quiz_attempts: dropping corrupt row "
+ "(score=%d outside [0, total=%d]) user=%s concept=%s",
+ score,
+ total,
+ user_id,
+ concept_node_id,
+ )
+ continue
+ accuracy = score / total
+ attempts.append(
+ RecentQuizAttempt(
+ score=score,
+ total=total,
+ difficulty=str(r.get("difficulty") or ""),
+ completed_at=r.get("completed_at"),
+ accuracy=round(accuracy, 4),
+ )
+ )
+
+ return QuizHistory(
+ summary=_coerce_summary(summary_raw),
+ recent_attempts=attempts,
+ )
+
+
+async def read_recent_quiz_attempts_tool(
+ ctx: RunContext[SaplingDeps],
+ concept_node_id: str,
+) -> QuizHistory:
+ """Returns this student's history on one concept: a `summary`
+ string digesting their prior mistakes (mine for distractor
+ inspiration) and `recent_attempts` — the last 5 completed quiz
+ attempts on this concept, newest first, with `accuracy` precomputed
+ so you can apply the adaptive-difficulty rule directly. Empty
+ history on first attempt. Pass the `concept_node_id` from the
+ user message; user identity is taken from context.
+ """
+ # user_id comes from ctx.deps so a tool call can't cross users.
+ return await read_recent_quiz_attempts(ctx.deps.user_id, concept_node_id)

backend/routes/quiz.py

@@ -166,11 +166,15 @@ async def _quiz_via_agent(
  supabase=None,
  request_id=request_id,
  )
+ # Keep this message routing-only; the workflow + adaptive rules
+ # live in the system prompt. We just hand the agent the inputs it
+ # needs and trust the prompt to drive tool calls.
  user_message = (
  f"Generate {num_questions} {difficulty} questions for the student. "
- f"The target concept is '{concept_name}' (concept_node_id={concept_node_id}). "
- f"Call read_concepts_for_user to find the student's weakest concepts in this course "
- f"and bias the question mix toward those."
+ f"The target concept is '{concept_name}' "
+ f"(concept_node_id={concept_node_id}). Follow the workflow in your "
+ f"system prompt; pass concept_node_id='{concept_node_id}' to "
+ f"read_recent_quiz_attempts."
  )
  if use_shared_context:
  user_message += (