feat(quiz): weight generation toward an approaching exam (#555)

backend/db/migrations/20260822090747_quiz_attempts_exam_days_away.sql

@@ -0,0 +1,29 @@
+-- H3 (#555): how many days away the student's next exam was when this quiz
+-- was generated.
+--
+-- Stored on the attempt, not merely used in the prompt, because the point of
+-- the issue is to be able to ASK LATER whether deadline-aware quizzes perform
+-- differently. A value that only ever reached a prompt leaves no way to
+-- compare a quiz taken three days before a midterm against one taken in week
+-- two — the measurement is the deliverable, not the prompt line.
+--
+-- Nullable with no default, and the distinction is load-bearing:
+-- NULL = we did not know (no course, no enrollment, no dated exams, or the
+-- lookup failed — `days_until_next_exam` degrades to None rather
+-- than failing a generation)
+-- 0 = the exam is TODAY, which is the most actionable value the feature
+-- produces. A DEFAULT 0 would make every legacy row and every
+-- unknown look like exam day.
+--
+-- Dates only. This column is a day count derived from `assignments.due_date`,
+-- which is plaintext and indexed; no grade VALUE is read, stored or prompted
+-- anywhere on this path (points columns are encrypted per #521, and the audit
+-- flags that the current ToS does not clearly cover feeding grades to a
+-- model).
+--
+-- Additive and idempotent. Ordering still matters: PostgREST 400s on a column
+-- its schema cache doesn't have, and `generate_quiz` includes this key in the
+-- attempt INSERT — so this must be applied strictly before the code ships, as
+-- with 20260814051517.
+ALTER TABLE quiz_attempts
+ ADD COLUMN IF NOT EXISTS exam_days_away INTEGER;

backend/routes/quiz.py

@@ -42,6 +42,7 @@
 from services.encryption import encrypt_json, decrypt_json_column
 from services.graph_service import apply_graph_update
 from services.quiz_context_service import get_quiz_context, save_quiz_context
+from services.exam_proximity import PROMPT_HORIZON_DAYS, days_until_next_exam
 from services.quiz_distractors import build_distractor_profile
 from services.fingerprint import fingerprint
 from services.quiz_identity import question_hash, normalize_text
@@ -795,6 +796,50 @@ def _do_not_repeat_block(recent: list[RecentQuestion]) -> str:
  )
 
 
+def _insert_attempt(attempt_row: dict) -> None:
+ """Write the attempt, surviving a schema that predates `exam_days_away`.
+
+ The omit-when-None rule alone does NOT make a pre-migration environment
+ safe, and the failure is nastier than it looks: it strikes exactly the
+ students the feature is FOR (the ones with a dated upcoming exam), so it
+ presents as a random partial outage rather than an obvious missing
+ migration. And it strikes late — the agent has already run and been
+ billed — so an unhandled 400 here loses the generated quiz, writes no
+ attempt row, emits no `quiz.generation_failed`, and never refunds the
+ rate-limit slot.
+ #
+ Ordering is still the rule (migration before code, as with
+ 20260814051517). This is the seatbelt for the window where PostgREST has
+ not yet reloaded its schema cache, not a licence to deploy first.
+ """
+ try:
+ table("quiz_attempts").insert(attempt_row)
+ return
+ except Exception:
+ if "exam_days_away" not in attempt_row:
+ raise
+ retry = {k: v for k, v in attempt_row.items() if k != "exam_days_away"}
+ logger.warning(
+ "quiz: attempt insert failed with exam_days_away present; retrying "
+ "without it (is 20260822090747 applied?) quiz_id=%s",
+ attempt_row.get("id"),
+ )
+ table("quiz_attempts").insert(retry)
+
+
+class GeneratedQuiz(NamedTuple):
+ """What one generation produced.
+
+ `exam_days_away` rides back with the questions rather than being resolved
+ again by the caller: it is used for the prompt here and stored on the
+ attempt there, and two lookups could disagree if an exam were entered
+ between them.
+ """
+
+ questions: list[dict]
+ exam_days_away: int | None = None
+
+
 async def _quiz_via_agent(
  *,
  user_id: str,
@@ -806,7 +851,7 @@ async def _quiz_via_agent(
  use_shared_context: bool,
  request_id: str,
  model_pref: str | None = None,
-) -> list[dict]:
+) -> GeneratedQuiz:
  """Run quiz_agent and return questions in the legacy wire shape.
 
  The agent's tools (read_concepts_for_user, read_misconceptions_for_course)
@@ -849,42 +894,39 @@ async def _quiz_via_agent(
  difficulty_clause = (
  f"Generate {num_questions} {difficulty} questions for the student."
  )
- routing_msg = (
- f"{difficulty_clause} "
- 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:
- routing_msg += (
- " Also call read_misconceptions_for_course and use those misconceptions "
- "as distractors and probes."
- )
 
  # Course-material grounding does blocking network I/O (a Gemini
  # embedding call, bounded at 60s) plus sync Supabase reads. Run it in a
  # worker thread so a slow/stalled retrieval can't freeze this worker's
  # event loop for every other in-flight request. Matches the
  # asyncio.to_thread pattern used by the agent read tools.
  #
- # E6's recently-asked read is an independent Supabase read + decrypt, so
- # it runs CONCURRENTLY with grounding rather than after it — the two have
- # nothing to say to each other and serializing them would add the slower
- # one's latency to every generation.
+ # E6's recently-asked read and H3's exam-proximity lookup are independent
+ # Supabase reads, so all three run CONCURRENTLY rather than in sequence —
+ # they have nothing to say to each other, and serializing them would add
+ # every one of their latencies to every generation. Proximity costs
+ # several round-trips on its own, so running it before this block made it
+ # fully additive.
  #
- # return_exceptions=True because BOTH are best-effort context, and a bare
- # gather propagates the first failure straight out of generation: an
+ # return_exceptions=True because ALL THREE are best-effort context, and a
+ # bare gather propagates the first failure straight out of generation: an
  # unreadable past attempt would 502 a quiz that needed no history at all.
  # Each helper already degrades internally; this is the backstop for the
  # failure they cannot catch (an unexpected raise on the way in or out).
- material, recent = await asyncio.gather(
+ material, recent, exam_days_away = await asyncio.gather(
  asyncio.to_thread(_course_material, course_id, concept_name),
  asyncio.to_thread(
  recent_question_identities, user_id, concept_node_id
  ),
+ asyncio.to_thread(days_until_next_exam, user_id, course_id),
  return_exceptions=True,
  )
+ if isinstance(exam_days_away, BaseException):
+ logger.warning(
+ "quiz: exam-proximity lookup failed (%s); generating without it",
+ type(exam_days_away).__name__, exc_info=exam_days_away,
+ )
+ exam_days_away = None
  if isinstance(material, BaseException):
  logger.warning(
  "quiz: course-material assembly failed (%s); generating ungrounded",
@@ -900,6 +942,40 @@ async def _quiz_via_agent(
  "do-not-repeat list", type(recent).__name__, exc_info=recent,
  )
  recent = []
+
+ routing_msg = (
+ f"{difficulty_clause} "
+ 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:
+ routing_msg += (
+ " Also call read_misconceptions_for_course and use those misconceptions "
+ "as distractors and probes."
+ )
+ # H3/#555: one line, dates only. Says how near the deadline is and lets
+ # the model decide what that implies — not "make it harder", which would
+ # contradict the adaptive difficulty the student actually chose. Omitted
+ # entirely when unknown: "next exam: unknown" is prompt tokens spent to
+ # say nothing.
+ # Bounded by a horizon: a final dated 87 days out would otherwise put
+ # "there is an exam coming, weight toward what an exam tests" on EVERY
+ # quiz for the whole semester, which carries no proximity signal and
+ # steers week-two practice toward exam-style questions — the opposite of
+ # what this line is for. The stored `exam_days_away` is NOT clamped: the
+ # analytics want the real distance, including the far ones.
+ if exam_days_away is not None and exam_days_away <= PROMPT_HORIZON_DAYS:
+ when = (
+ "TODAY" if exam_days_away == 0
+ else "tomorrow" if exam_days_away == 1
+ else f"in {exam_days_away} days"
+ )
+ routing_msg += (
+ f" The student's next exam in this course is {when}. Weight the"
+ " questions toward what an exam would actually test."
+ )
  _log_rag_uncovered(
  material,
  user_id=user_id,
@@ -1078,7 +1154,10 @@ def _absorb(quiz: Quiz, model: str) -> None:
  "quiz_agent produced no valid questions after wire-format validation"
  )
  # Never serve more than asked for (a generous top-up run can overshoot).
- return wire_questions[:num_questions]
+ return GeneratedQuiz(
+ questions=wire_questions[:num_questions],
+ exam_days_away=exam_days_away,
+ )
 
 @router.get("/config")
 def quiz_config():
@@ -1169,7 +1248,7 @@ async def generate_quiz(body: GenerateQuizBody, request: Request):
  # QUIZ_GENERATION_TIMEOUT_SEC (see _run) — cancelling the whole
  # coroutine here would discard a partial quiz the top-up handler
  # is designed to serve.
-questions = await _quiz_via_agent(
+generated = await _quiz_via_agent(
  user_id=body.user_id,
  course_id=course_id,
  concept_node_id=body.concept_node_id,
@@ -1180,6 +1259,8 @@ async def generate_quiz(body: GenerateQuizBody, request: Request):
  request_id=request_id,
  model_pref=body.model_pref,
  )
+ questions = generated.questions
+ exam_days_away = generated.exam_days_away
  except HTTPException:
  # The 404 for an unknown concept node is raised before the agent call;
  # never swallow a known HTTP state.
@@ -1224,13 +1305,21 @@ async def generate_quiz(body: GenerateQuizBody, request: Request):
  ) from e
 
  quiz_id = str(uuid.uuid4())
-table("quiz_attempts").insert({
+attempt_row = {
  "id": quiz_id,
  "user_id": body.user_id,
  "concept_node_id": body.concept_node_id,
  "difficulty": body.difficulty,
  "questions_json": encrypt_json(questions),
- })
+ }
+ # H3/#555: recorded so the question "do deadline-aware quizzes perform
+ # differently?" is answerable later. Omitted when unknown rather than
+ # written as an explicit null, mirroring apply_graph_update's rule: an
+ # environment that took this code before the migration keeps generating
+ # instead of 400ing on a column PostgREST's schema cache doesn't have.
+ if exam_days_away is not None:
+ attempt_row["exam_days_away"] = exam_days_away
+ _insert_attempt(attempt_row)
  # #117: quiz.started once the attempt row exists. num_questions is the
  # actual generated count (the agent may return fewer than requested).
  events_service.log_event(

backend/routes/study_guide.py

@@ -25,6 +25,7 @@
 )
 from services.graph_service import get_courses as graph_get_courses
 from services.auth_guard import require_self
+from services.exam_proximity import is_exam
 from services.encryption import (
  decrypt_if_present,
  decrypt_json,
@@ -303,15 +304,11 @@ def get_exams(
  order="due_date.asc",
  ) or []
 
- exam_keywords = ["exam", "midterm", "final", "quiz"]
- exams = []
- for a in all_assignments:
- atype = (a.get("assignment_type") or "").lower()
- title = (a.get("title") or "").lower()
- if atype == "exam" or any(kw in title for kw in exam_keywords):
- exams.append(a)
-
- return {"exams": exams}
+ # The heuristic moved to services/exam_proximity.py (#555), which needed
+ # the same question answered on the quiz path. One definition, two
+ # callers — a second copy would drift, which is the failure #557 spent a
+ # workstream undoing.
+ return {"exams": [a for a in all_assignments if is_exam(a)]}
 
 
 @router.get("/{user_id}/guide")

backend/scripts/benchmark_quiz.py

@@ -147,7 +147,10 @@ async def _generate_quiz_for_async(concept_name: str) -> list[dict]:
  so this exercises the real production grounding path rather than
  generating an ungrounded quiz.
  """
- return await _quiz_via_agent(
+ # `.questions`: _quiz_via_agent returns a GeneratedQuiz (#555) so the
+ # exam-proximity value it resolved can reach the attempt row. This bench
+ # only wants the questions.
+ generated = await _quiz_via_agent(
  user_id="quizfix-user-0001",
  course_id=FIXTURE_COURSE_ID,
  concept_node_id="quizfix-node-0001",
@@ -157,6 +160,7 @@ async def _generate_quiz_for_async(concept_name: str) -> list[dict]:
  use_shared_context=False,
  request_id="quizfix-bench",
  )
+ return generated.questions
 
 
 def generate_quiz_for(concept_name: str) -> list[dict]:

backend/db/migrations/20260822090747_quiz_attempts_exam_days_away.sql

@@ -0,0 +1,29 @@
+-- H3 (#555): how many days away the student's next exam was when this quiz
+-- was generated.
+--
+-- Stored on the attempt, not merely used in the prompt, because the point of
+-- the issue is to be able to ASK LATER whether deadline-aware quizzes perform
+-- differently. A value that only ever reached a prompt leaves no way to
+-- compare a quiz taken three days before a midterm against one taken in week
+-- two — the measurement is the deliverable, not the prompt line.
+--
+-- Nullable with no default, and the distinction is load-bearing:
+-- NULL = we did not know (no course, no enrollment, no dated exams, or the
+-- lookup failed — `days_until_next_exam` degrades to None rather
+-- than failing a generation)
+-- 0 = the exam is TODAY, which is the most actionable value the feature
+-- produces. A DEFAULT 0 would make every legacy row and every
+-- unknown look like exam day.
+--
+-- Dates only. This column is a day count derived from `assignments.due_date`,
+-- which is plaintext and indexed; no grade VALUE is read, stored or prompted
+-- anywhere on this path (points columns are encrypted per #521, and the audit
+-- flags that the current ToS does not clearly cover feeding grades to a
+-- model).
+--
+-- Additive and idempotent. Ordering still matters: PostgREST 400s on a column
+-- its schema cache doesn't have, and `generate_quiz` includes this key in the
+-- attempt INSERT — so this must be applied strictly before the code ships, as
+-- with 20260814051517.
+ALTER TABLE quiz_attempts
+ ADD COLUMN IF NOT EXISTS exam_days_away INTEGER;

backend/routes/quiz.py

@@ -42,6 +42,7 @@
 from services.encryption import encrypt_json, decrypt_json_column
 from services.graph_service import apply_graph_update
 from services.quiz_context_service import get_quiz_context, save_quiz_context
+from services.exam_proximity import PROMPT_HORIZON_DAYS, days_until_next_exam
 from services.quiz_distractors import build_distractor_profile
 from services.fingerprint import fingerprint
 from services.quiz_identity import question_hash, normalize_text
@@ -795,6 +796,50 @@ def _do_not_repeat_block(recent: list[RecentQuestion]) -> str:
  )
 
 
+def _insert_attempt(attempt_row: dict) -> None:
+ """Write the attempt, surviving a schema that predates `exam_days_away`.
+
+ The omit-when-None rule alone does NOT make a pre-migration environment
+ safe, and the failure is nastier than it looks: it strikes exactly the
+ students the feature is FOR (the ones with a dated upcoming exam), so it
+ presents as a random partial outage rather than an obvious missing
+ migration. And it strikes late — the agent has already run and been
+ billed — so an unhandled 400 here loses the generated quiz, writes no
+ attempt row, emits no `quiz.generation_failed`, and never refunds the
+ rate-limit slot.
+ #
+ Ordering is still the rule (migration before code, as with
+ 20260814051517). This is the seatbelt for the window where PostgREST has
+ not yet reloaded its schema cache, not a licence to deploy first.
+ """
+ try:
+ table("quiz_attempts").insert(attempt_row)
+ return
+ except Exception:
+ if "exam_days_away" not in attempt_row:
+ raise
+ retry = {k: v for k, v in attempt_row.items() if k != "exam_days_away"}
+ logger.warning(
+ "quiz: attempt insert failed with exam_days_away present; retrying "
+ "without it (is 20260822090747 applied?) quiz_id=%s",
+ attempt_row.get("id"),
+ )
+ table("quiz_attempts").insert(retry)
+
+
+class GeneratedQuiz(NamedTuple):
+ """What one generation produced.
+
+ `exam_days_away` rides back with the questions rather than being resolved
+ again by the caller: it is used for the prompt here and stored on the
+ attempt there, and two lookups could disagree if an exam were entered
+ between them.
+ """
+
+ questions: list[dict]
+ exam_days_away: int | None = None
+
+
 async def _quiz_via_agent(
  *,
  user_id: str,
@@ -806,7 +851,7 @@ async def _quiz_via_agent(
  use_shared_context: bool,
  request_id: str,
  model_pref: str | None = None,
-) -> list[dict]:
+) -> GeneratedQuiz:
  """Run quiz_agent and return questions in the legacy wire shape.
 
  The agent's tools (read_concepts_for_user, read_misconceptions_for_course)
@@ -849,42 +894,39 @@ async def _quiz_via_agent(
  difficulty_clause = (
  f"Generate {num_questions} {difficulty} questions for the student."
  )
- routing_msg = (
- f"{difficulty_clause} "
- 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:
- routing_msg += (
- " Also call read_misconceptions_for_course and use those misconceptions "
- "as distractors and probes."
- )
 
  # Course-material grounding does blocking network I/O (a Gemini
  # embedding call, bounded at 60s) plus sync Supabase reads. Run it in a
  # worker thread so a slow/stalled retrieval can't freeze this worker's
  # event loop for every other in-flight request. Matches the
  # asyncio.to_thread pattern used by the agent read tools.
  #
- # E6's recently-asked read is an independent Supabase read + decrypt, so
- # it runs CONCURRENTLY with grounding rather than after it — the two have
- # nothing to say to each other and serializing them would add the slower
- # one's latency to every generation.
+ # E6's recently-asked read and H3's exam-proximity lookup are independent
+ # Supabase reads, so all three run CONCURRENTLY rather than in sequence —
+ # they have nothing to say to each other, and serializing them would add
+ # every one of their latencies to every generation. Proximity costs
+ # several round-trips on its own, so running it before this block made it
+ # fully additive.
  #
- # return_exceptions=True because BOTH are best-effort context, and a bare
- # gather propagates the first failure straight out of generation: an
+ # return_exceptions=True because ALL THREE are best-effort context, and a
+ # bare gather propagates the first failure straight out of generation: an
  # unreadable past attempt would 502 a quiz that needed no history at all.
  # Each helper already degrades internally; this is the backstop for the
  # failure they cannot catch (an unexpected raise on the way in or out).
- material, recent = await asyncio.gather(
+ material, recent, exam_days_away = await asyncio.gather(
  asyncio.to_thread(_course_material, course_id, concept_name),
  asyncio.to_thread(
  recent_question_identities, user_id, concept_node_id
  ),
+ asyncio.to_thread(days_until_next_exam, user_id, course_id),
  return_exceptions=True,
  )
+ if isinstance(exam_days_away, BaseException):
+ logger.warning(
+ "quiz: exam-proximity lookup failed (%s); generating without it",
+ type(exam_days_away).__name__, exc_info=exam_days_away,
+ )
+ exam_days_away = None
  if isinstance(material, BaseException):
  logger.warning(
  "quiz: course-material assembly failed (%s); generating ungrounded",
@@ -900,6 +942,40 @@ async def _quiz_via_agent(
  "do-not-repeat list", type(recent).__name__, exc_info=recent,
  )
  recent = []
+
+ routing_msg = (
+ f"{difficulty_clause} "
+ 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:
+ routing_msg += (
+ " Also call read_misconceptions_for_course and use those misconceptions "
+ "as distractors and probes."
+ )
+ # H3/#555: one line, dates only. Says how near the deadline is and lets
+ # the model decide what that implies — not "make it harder", which would
+ # contradict the adaptive difficulty the student actually chose. Omitted
+ # entirely when unknown: "next exam: unknown" is prompt tokens spent to
+ # say nothing.
+ # Bounded by a horizon: a final dated 87 days out would otherwise put
+ # "there is an exam coming, weight toward what an exam tests" on EVERY
+ # quiz for the whole semester, which carries no proximity signal and
+ # steers week-two practice toward exam-style questions — the opposite of
+ # what this line is for. The stored `exam_days_away` is NOT clamped: the
+ # analytics want the real distance, including the far ones.
+ if exam_days_away is not None and exam_days_away <= PROMPT_HORIZON_DAYS:
+ when = (
+ "TODAY" if exam_days_away == 0
+ else "tomorrow" if exam_days_away == 1
+ else f"in {exam_days_away} days"
+ )
+ routing_msg += (
+ f" The student's next exam in this course is {when}. Weight the"
+ " questions toward what an exam would actually test."
+ )
  _log_rag_uncovered(
  material,
  user_id=user_id,
@@ -1078,7 +1154,10 @@ def _absorb(quiz: Quiz, model: str) -> None:
  "quiz_agent produced no valid questions after wire-format validation"
  )
  # Never serve more than asked for (a generous top-up run can overshoot).
- return wire_questions[:num_questions]
+ return GeneratedQuiz(
+ questions=wire_questions[:num_questions],
+ exam_days_away=exam_days_away,
+ )
 
 @router.get("/config")
 def quiz_config():
@@ -1169,7 +1248,7 @@ async def generate_quiz(body: GenerateQuizBody, request: Request):
  # QUIZ_GENERATION_TIMEOUT_SEC (see _run) — cancelling the whole
  # coroutine here would discard a partial quiz the top-up handler
  # is designed to serve.
-questions = await _quiz_via_agent(
+generated = await _quiz_via_agent(
  user_id=body.user_id,
  course_id=course_id,
  concept_node_id=body.concept_node_id,
@@ -1180,6 +1259,8 @@ async def generate_quiz(body: GenerateQuizBody, request: Request):
  request_id=request_id,
  model_pref=body.model_pref,
  )
+ questions = generated.questions
+ exam_days_away = generated.exam_days_away
  except HTTPException:
  # The 404 for an unknown concept node is raised before the agent call;
  # never swallow a known HTTP state.
@@ -1224,13 +1305,21 @@ async def generate_quiz(body: GenerateQuizBody, request: Request):
  ) from e
 
  quiz_id = str(uuid.uuid4())
-table("quiz_attempts").insert({
+attempt_row = {
  "id": quiz_id,
  "user_id": body.user_id,
  "concept_node_id": body.concept_node_id,
  "difficulty": body.difficulty,
  "questions_json": encrypt_json(questions),
- })
+ }
+ # H3/#555: recorded so the question "do deadline-aware quizzes perform
+ # differently?" is answerable later. Omitted when unknown rather than
+ # written as an explicit null, mirroring apply_graph_update's rule: an
+ # environment that took this code before the migration keeps generating
+ # instead of 400ing on a column PostgREST's schema cache doesn't have.
+ if exam_days_away is not None:
+ attempt_row["exam_days_away"] = exam_days_away
+ _insert_attempt(attempt_row)
  # #117: quiz.started once the attempt row exists. num_questions is the
  # actual generated count (the agent may return fewer than requested).
  events_service.log_event(

backend/routes/study_guide.py

@@ -25,6 +25,7 @@
 )
 from services.graph_service import get_courses as graph_get_courses
 from services.auth_guard import require_self
+from services.exam_proximity import is_exam
 from services.encryption import (
  decrypt_if_present,
  decrypt_json,
@@ -303,15 +304,11 @@ def get_exams(
  order="due_date.asc",
  ) or []
 
- exam_keywords = ["exam", "midterm", "final", "quiz"]
- exams = []
- for a in all_assignments:
- atype = (a.get("assignment_type") or "").lower()
- title = (a.get("title") or "").lower()
- if atype == "exam" or any(kw in title for kw in exam_keywords):
- exams.append(a)
-
- return {"exams": exams}
+ # The heuristic moved to services/exam_proximity.py (#555), which needed
+ # the same question answered on the quiz path. One definition, two
+ # callers — a second copy would drift, which is the failure #557 spent a
+ # workstream undoing.
+ return {"exams": [a for a in all_assignments if is_exam(a)]}
 
 
 @router.get("/{user_id}/guide")

backend/scripts/benchmark_quiz.py

@@ -147,7 +147,10 @@ async def _generate_quiz_for_async(concept_name: str) -> list[dict]:
  so this exercises the real production grounding path rather than
  generating an ungrounded quiz.
  """
- return await _quiz_via_agent(
+ # `.questions`: _quiz_via_agent returns a GeneratedQuiz (#555) so the
+ # exam-proximity value it resolved can reach the attempt row. This bench
+ # only wants the questions.
+ generated = await _quiz_via_agent(
  user_id="quizfix-user-0001",
  course_id=FIXTURE_COURSE_ID,
  concept_node_id="quizfix-node-0001",
@@ -157,6 +160,7 @@ async def _generate_quiz_for_async(concept_name: str) -> list[dict]:
  use_shared_context=False,
  request_id="quizfix-bench",
  )
+ return generated.questions
 
 
 def generate_quiz_for(concept_name: str) -> list[dict]:

backend/db/migrations/20260822090747_quiz_attempts_exam_days_away.sql

@@ -0,0 +1,29 @@
+-- H3 (#555): how many days away the student's next exam was when this quiz
+-- was generated.
+--
+-- Stored on the attempt, not merely used in the prompt, because the point of
+-- the issue is to be able to ASK LATER whether deadline-aware quizzes perform
+-- differently. A value that only ever reached a prompt leaves no way to
+-- compare a quiz taken three days before a midterm against one taken in week
+-- two — the measurement is the deliverable, not the prompt line.
+--
+-- Nullable with no default, and the distinction is load-bearing:
+-- NULL = we did not know (no course, no enrollment, no dated exams, or the
+-- lookup failed — `days_until_next_exam` degrades to None rather
+-- than failing a generation)
+-- 0 = the exam is TODAY, which is the most actionable value the feature
+-- produces. A DEFAULT 0 would make every legacy row and every
+-- unknown look like exam day.
+--
+-- Dates only. This column is a day count derived from `assignments.due_date`,
+-- which is plaintext and indexed; no grade VALUE is read, stored or prompted
+-- anywhere on this path (points columns are encrypted per #521, and the audit
+-- flags that the current ToS does not clearly cover feeding grades to a
+-- model).
+--
+-- Additive and idempotent. Ordering still matters: PostgREST 400s on a column
+-- its schema cache doesn't have, and `generate_quiz` includes this key in the
+-- attempt INSERT — so this must be applied strictly before the code ships, as
+-- with 20260814051517.
+ALTER TABLE quiz_attempts
+ ADD COLUMN IF NOT EXISTS exam_days_away INTEGER;

backend/routes/quiz.py

@@ -42,6 +42,7 @@
 from services.encryption import encrypt_json, decrypt_json_column
 from services.graph_service import apply_graph_update
 from services.quiz_context_service import get_quiz_context, save_quiz_context
+from services.exam_proximity import PROMPT_HORIZON_DAYS, days_until_next_exam
 from services.quiz_distractors import build_distractor_profile
 from services.fingerprint import fingerprint
 from services.quiz_identity import question_hash, normalize_text
@@ -795,6 +796,50 @@ def _do_not_repeat_block(recent: list[RecentQuestion]) -> str:
  )
 
 
+def _insert_attempt(attempt_row: dict) -> None:
+ """Write the attempt, surviving a schema that predates `exam_days_away`.
+
+ The omit-when-None rule alone does NOT make a pre-migration environment
+ safe, and the failure is nastier than it looks: it strikes exactly the
+ students the feature is FOR (the ones with a dated upcoming exam), so it
+ presents as a random partial outage rather than an obvious missing
+ migration. And it strikes late — the agent has already run and been
+ billed — so an unhandled 400 here loses the generated quiz, writes no
+ attempt row, emits no `quiz.generation_failed`, and never refunds the
+ rate-limit slot.
+ #
+ Ordering is still the rule (migration before code, as with
+ 20260814051517). This is the seatbelt for the window where PostgREST has
+ not yet reloaded its schema cache, not a licence to deploy first.
+ """
+ try:
+ table("quiz_attempts").insert(attempt_row)
+ return
+ except Exception:
+ if "exam_days_away" not in attempt_row:
+ raise
+ retry = {k: v for k, v in attempt_row.items() if k != "exam_days_away"}
+ logger.warning(
+ "quiz: attempt insert failed with exam_days_away present; retrying "
+ "without it (is 20260822090747 applied?) quiz_id=%s",
+ attempt_row.get("id"),
+ )
+ table("quiz_attempts").insert(retry)
+
+
+class GeneratedQuiz(NamedTuple):
+ """What one generation produced.
+
+ `exam_days_away` rides back with the questions rather than being resolved
+ again by the caller: it is used for the prompt here and stored on the
+ attempt there, and two lookups could disagree if an exam were entered
+ between them.
+ """
+
+ questions: list[dict]
+ exam_days_away: int | None = None
+
+
 async def _quiz_via_agent(
  *,
  user_id: str,
@@ -806,7 +851,7 @@ async def _quiz_via_agent(
  use_shared_context: bool,
  request_id: str,
  model_pref: str | None = None,
-) -> list[dict]:
+) -> GeneratedQuiz:
  """Run quiz_agent and return questions in the legacy wire shape.
 
  The agent's tools (read_concepts_for_user, read_misconceptions_for_course)
@@ -849,42 +894,39 @@ async def _quiz_via_agent(
  difficulty_clause = (
  f"Generate {num_questions} {difficulty} questions for the student."
  )
- routing_msg = (
- f"{difficulty_clause} "
- 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:
- routing_msg += (
- " Also call read_misconceptions_for_course and use those misconceptions "
- "as distractors and probes."
- )
 
  # Course-material grounding does blocking network I/O (a Gemini
  # embedding call, bounded at 60s) plus sync Supabase reads. Run it in a
  # worker thread so a slow/stalled retrieval can't freeze this worker's
  # event loop for every other in-flight request. Matches the
  # asyncio.to_thread pattern used by the agent read tools.
  #
- # E6's recently-asked read is an independent Supabase read + decrypt, so
- # it runs CONCURRENTLY with grounding rather than after it — the two have
- # nothing to say to each other and serializing them would add the slower
- # one's latency to every generation.
+ # E6's recently-asked read and H3's exam-proximity lookup are independent
+ # Supabase reads, so all three run CONCURRENTLY rather than in sequence —
+ # they have nothing to say to each other, and serializing them would add
+ # every one of their latencies to every generation. Proximity costs
+ # several round-trips on its own, so running it before this block made it
+ # fully additive.
  #
- # return_exceptions=True because BOTH are best-effort context, and a bare
- # gather propagates the first failure straight out of generation: an
+ # return_exceptions=True because ALL THREE are best-effort context, and a
+ # bare gather propagates the first failure straight out of generation: an
  # unreadable past attempt would 502 a quiz that needed no history at all.
  # Each helper already degrades internally; this is the backstop for the
  # failure they cannot catch (an unexpected raise on the way in or out).
- material, recent = await asyncio.gather(
+ material, recent, exam_days_away = await asyncio.gather(
  asyncio.to_thread(_course_material, course_id, concept_name),
  asyncio.to_thread(
  recent_question_identities, user_id, concept_node_id
  ),
+ asyncio.to_thread(days_until_next_exam, user_id, course_id),
  return_exceptions=True,
  )
+ if isinstance(exam_days_away, BaseException):
+ logger.warning(
+ "quiz: exam-proximity lookup failed (%s); generating without it",
+ type(exam_days_away).__name__, exc_info=exam_days_away,
+ )
+ exam_days_away = None
  if isinstance(material, BaseException):
  logger.warning(
  "quiz: course-material assembly failed (%s); generating ungrounded",
@@ -900,6 +942,40 @@ async def _quiz_via_agent(
  "do-not-repeat list", type(recent).__name__, exc_info=recent,
  )
  recent = []
+
+ routing_msg = (
+ f"{difficulty_clause} "
+ 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:
+ routing_msg += (
+ " Also call read_misconceptions_for_course and use those misconceptions "
+ "as distractors and probes."
+ )
+ # H3/#555: one line, dates only. Says how near the deadline is and lets
+ # the model decide what that implies — not "make it harder", which would
+ # contradict the adaptive difficulty the student actually chose. Omitted
+ # entirely when unknown: "next exam: unknown" is prompt tokens spent to
+ # say nothing.
+ # Bounded by a horizon: a final dated 87 days out would otherwise put
+ # "there is an exam coming, weight toward what an exam tests" on EVERY
+ # quiz for the whole semester, which carries no proximity signal and
+ # steers week-two practice toward exam-style questions — the opposite of
+ # what this line is for. The stored `exam_days_away` is NOT clamped: the
+ # analytics want the real distance, including the far ones.
+ if exam_days_away is not None and exam_days_away <= PROMPT_HORIZON_DAYS:
+ when = (
+ "TODAY" if exam_days_away == 0
+ else "tomorrow" if exam_days_away == 1
+ else f"in {exam_days_away} days"
+ )
+ routing_msg += (
+ f" The student's next exam in this course is {when}. Weight the"
+ " questions toward what an exam would actually test."
+ )
  _log_rag_uncovered(
  material,
  user_id=user_id,
@@ -1078,7 +1154,10 @@ def _absorb(quiz: Quiz, model: str) -> None:
  "quiz_agent produced no valid questions after wire-format validation"
  )
  # Never serve more than asked for (a generous top-up run can overshoot).
- return wire_questions[:num_questions]
+ return GeneratedQuiz(
+ questions=wire_questions[:num_questions],
+ exam_days_away=exam_days_away,
+ )
 
 @router.get("/config")
 def quiz_config():
@@ -1169,7 +1248,7 @@ async def generate_quiz(body: GenerateQuizBody, request: Request):
  # QUIZ_GENERATION_TIMEOUT_SEC (see _run) — cancelling the whole
  # coroutine here would discard a partial quiz the top-up handler
  # is designed to serve.
-questions = await _quiz_via_agent(
+generated = await _quiz_via_agent(
  user_id=body.user_id,
  course_id=course_id,
  concept_node_id=body.concept_node_id,
@@ -1180,6 +1259,8 @@ async def generate_quiz(body: GenerateQuizBody, request: Request):
  request_id=request_id,
  model_pref=body.model_pref,
  )
+ questions = generated.questions
+ exam_days_away = generated.exam_days_away
  except HTTPException:
  # The 404 for an unknown concept node is raised before the agent call;
  # never swallow a known HTTP state.
@@ -1224,13 +1305,21 @@ async def generate_quiz(body: GenerateQuizBody, request: Request):
  ) from e
 
  quiz_id = str(uuid.uuid4())
-table("quiz_attempts").insert({
+attempt_row = {
  "id": quiz_id,
  "user_id": body.user_id,
  "concept_node_id": body.concept_node_id,
  "difficulty": body.difficulty,
  "questions_json": encrypt_json(questions),
- })
+ }
+ # H3/#555: recorded so the question "do deadline-aware quizzes perform
+ # differently?" is answerable later. Omitted when unknown rather than
+ # written as an explicit null, mirroring apply_graph_update's rule: an
+ # environment that took this code before the migration keeps generating
+ # instead of 400ing on a column PostgREST's schema cache doesn't have.
+ if exam_days_away is not None:
+ attempt_row["exam_days_away"] = exam_days_away
+ _insert_attempt(attempt_row)
  # #117: quiz.started once the attempt row exists. num_questions is the
  # actual generated count (the agent may return fewer than requested).
  events_service.log_event(

backend/routes/study_guide.py

@@ -25,6 +25,7 @@
 )
 from services.graph_service import get_courses as graph_get_courses
 from services.auth_guard import require_self
+from services.exam_proximity import is_exam
 from services.encryption import (
  decrypt_if_present,
  decrypt_json,
@@ -303,15 +304,11 @@ def get_exams(
  order="due_date.asc",
  ) or []
 
- exam_keywords = ["exam", "midterm", "final", "quiz"]
- exams = []
- for a in all_assignments:
- atype = (a.get("assignment_type") or "").lower()
- title = (a.get("title") or "").lower()
- if atype == "exam" or any(kw in title for kw in exam_keywords):
- exams.append(a)
-
- return {"exams": exams}
+ # The heuristic moved to services/exam_proximity.py (#555), which needed
+ # the same question answered on the quiz path. One definition, two
+ # callers — a second copy would drift, which is the failure #557 spent a
+ # workstream undoing.
+ return {"exams": [a for a in all_assignments if is_exam(a)]}
 
 
 @router.get("/{user_id}/guide")

backend/scripts/benchmark_quiz.py

@@ -147,7 +147,10 @@ async def _generate_quiz_for_async(concept_name: str) -> list[dict]:
  so this exercises the real production grounding path rather than
  generating an ungrounded quiz.
  """
- return await _quiz_via_agent(
+ # `.questions`: _quiz_via_agent returns a GeneratedQuiz (#555) so the
+ # exam-proximity value it resolved can reach the attempt row. This bench
+ # only wants the questions.
+ generated = await _quiz_via_agent(
  user_id="quizfix-user-0001",
  course_id=FIXTURE_COURSE_ID,
  concept_node_id="quizfix-node-0001",
@@ -157,6 +160,7 @@ async def _generate_quiz_for_async(concept_name: str) -> list[dict]:
  use_shared_context=False,
  request_id="quizfix-bench",
  )
+ return generated.questions
 
 
 def generate_quiz_for(concept_name: str) -> list[dict]:

backend/db/migrations/20260822090747_quiz_attempts_exam_days_away.sql

@@ -0,0 +1,29 @@
+-- H3 (#555): how many days away the student's next exam was when this quiz
+-- was generated.
+--
+-- Stored on the attempt, not merely used in the prompt, because the point of
+-- the issue is to be able to ASK LATER whether deadline-aware quizzes perform
+-- differently. A value that only ever reached a prompt leaves no way to
+-- compare a quiz taken three days before a midterm against one taken in week
+-- two — the measurement is the deliverable, not the prompt line.
+--
+-- Nullable with no default, and the distinction is load-bearing:
+-- NULL = we did not know (no course, no enrollment, no dated exams, or the
+-- lookup failed — `days_until_next_exam` degrades to None rather
+-- than failing a generation)
+-- 0 = the exam is TODAY, which is the most actionable value the feature
+-- produces. A DEFAULT 0 would make every legacy row and every
+-- unknown look like exam day.
+--
+-- Dates only. This column is a day count derived from `assignments.due_date`,
+-- which is plaintext and indexed; no grade VALUE is read, stored or prompted
+-- anywhere on this path (points columns are encrypted per #521, and the audit
+-- flags that the current ToS does not clearly cover feeding grades to a
+-- model).
+--
+-- Additive and idempotent. Ordering still matters: PostgREST 400s on a column
+-- its schema cache doesn't have, and `generate_quiz` includes this key in the
+-- attempt INSERT — so this must be applied strictly before the code ships, as
+-- with 20260814051517.
+ALTER TABLE quiz_attempts
+ ADD COLUMN IF NOT EXISTS exam_days_away INTEGER;

backend/routes/quiz.py

@@ -42,6 +42,7 @@
 from services.encryption import encrypt_json, decrypt_json_column
 from services.graph_service import apply_graph_update
 from services.quiz_context_service import get_quiz_context, save_quiz_context
+from services.exam_proximity import PROMPT_HORIZON_DAYS, days_until_next_exam
 from services.quiz_distractors import build_distractor_profile
 from services.fingerprint import fingerprint
 from services.quiz_identity import question_hash, normalize_text
@@ -795,6 +796,50 @@ def _do_not_repeat_block(recent: list[RecentQuestion]) -> str:
  )
 
 
+def _insert_attempt(attempt_row: dict) -> None:
+ """Write the attempt, surviving a schema that predates `exam_days_away`.
+
+ The omit-when-None rule alone does NOT make a pre-migration environment
+ safe, and the failure is nastier than it looks: it strikes exactly the
+ students the feature is FOR (the ones with a dated upcoming exam), so it
+ presents as a random partial outage rather than an obvious missing
+ migration. And it strikes late — the agent has already run and been
+ billed — so an unhandled 400 here loses the generated quiz, writes no
+ attempt row, emits no `quiz.generation_failed`, and never refunds the
+ rate-limit slot.
+ #
+ Ordering is still the rule (migration before code, as with
+ 20260814051517). This is the seatbelt for the window where PostgREST has
+ not yet reloaded its schema cache, not a licence to deploy first.
+ """
+ try:
+ table("quiz_attempts").insert(attempt_row)
+ return
+ except Exception:
+ if "exam_days_away" not in attempt_row:
+ raise
+ retry = {k: v for k, v in attempt_row.items() if k != "exam_days_away"}
+ logger.warning(
+ "quiz: attempt insert failed with exam_days_away present; retrying "
+ "without it (is 20260822090747 applied?) quiz_id=%s",
+ attempt_row.get("id"),
+ )
+ table("quiz_attempts").insert(retry)
+
+
+class GeneratedQuiz(NamedTuple):
+ """What one generation produced.
+
+ `exam_days_away` rides back with the questions rather than being resolved
+ again by the caller: it is used for the prompt here and stored on the
+ attempt there, and two lookups could disagree if an exam were entered
+ between them.
+ """
+
+ questions: list[dict]
+ exam_days_away: int | None = None
+
+
 async def _quiz_via_agent(
  *,
  user_id: str,
@@ -806,7 +851,7 @@ async def _quiz_via_agent(
  use_shared_context: bool,
  request_id: str,
  model_pref: str | None = None,
-) -> list[dict]:
+) -> GeneratedQuiz:
  """Run quiz_agent and return questions in the legacy wire shape.
 
  The agent's tools (read_concepts_for_user, read_misconceptions_for_course)
@@ -849,42 +894,39 @@ async def _quiz_via_agent(
  difficulty_clause = (
  f"Generate {num_questions} {difficulty} questions for the student."
  )
- routing_msg = (
- f"{difficulty_clause} "
- 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:
- routing_msg += (
- " Also call read_misconceptions_for_course and use those misconceptions "
- "as distractors and probes."
- )
 
  # Course-material grounding does blocking network I/O (a Gemini
  # embedding call, bounded at 60s) plus sync Supabase reads. Run it in a
  # worker thread so a slow/stalled retrieval can't freeze this worker's
  # event loop for every other in-flight request. Matches the
  # asyncio.to_thread pattern used by the agent read tools.
  #
- # E6's recently-asked read is an independent Supabase read + decrypt, so
- # it runs CONCURRENTLY with grounding rather than after it — the two have
- # nothing to say to each other and serializing them would add the slower
- # one's latency to every generation.
+ # E6's recently-asked read and H3's exam-proximity lookup are independent
+ # Supabase reads, so all three run CONCURRENTLY rather than in sequence —
+ # they have nothing to say to each other, and serializing them would add
+ # every one of their latencies to every generation. Proximity costs
+ # several round-trips on its own, so running it before this block made it
+ # fully additive.
  #
- # return_exceptions=True because BOTH are best-effort context, and a bare
- # gather propagates the first failure straight out of generation: an
+ # return_exceptions=True because ALL THREE are best-effort context, and a
+ # bare gather propagates the first failure straight out of generation: an
  # unreadable past attempt would 502 a quiz that needed no history at all.
  # Each helper already degrades internally; this is the backstop for the
  # failure they cannot catch (an unexpected raise on the way in or out).
- material, recent = await asyncio.gather(
+ material, recent, exam_days_away = await asyncio.gather(
  asyncio.to_thread(_course_material, course_id, concept_name),
  asyncio.to_thread(
  recent_question_identities, user_id, concept_node_id
  ),
+ asyncio.to_thread(days_until_next_exam, user_id, course_id),
  return_exceptions=True,
  )
+ if isinstance(exam_days_away, BaseException):
+ logger.warning(
+ "quiz: exam-proximity lookup failed (%s); generating without it",
+ type(exam_days_away).__name__, exc_info=exam_days_away,
+ )
+ exam_days_away = None
  if isinstance(material, BaseException):
  logger.warning(
  "quiz: course-material assembly failed (%s); generating ungrounded",
@@ -900,6 +942,40 @@ async def _quiz_via_agent(
  "do-not-repeat list", type(recent).__name__, exc_info=recent,
  )
  recent = []
+
+ routing_msg = (
+ f"{difficulty_clause} "
+ 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:
+ routing_msg += (
+ " Also call read_misconceptions_for_course and use those misconceptions "
+ "as distractors and probes."
+ )
+ # H3/#555: one line, dates only. Says how near the deadline is and lets
+ # the model decide what that implies — not "make it harder", which would
+ # contradict the adaptive difficulty the student actually chose. Omitted
+ # entirely when unknown: "next exam: unknown" is prompt tokens spent to
+ # say nothing.
+ # Bounded by a horizon: a final dated 87 days out would otherwise put
+ # "there is an exam coming, weight toward what an exam tests" on EVERY
+ # quiz for the whole semester, which carries no proximity signal and
+ # steers week-two practice toward exam-style questions — the opposite of
+ # what this line is for. The stored `exam_days_away` is NOT clamped: the
+ # analytics want the real distance, including the far ones.
+ if exam_days_away is not None and exam_days_away <= PROMPT_HORIZON_DAYS:
+ when = (
+ "TODAY" if exam_days_away == 0
+ else "tomorrow" if exam_days_away == 1
+ else f"in {exam_days_away} days"
+ )
+ routing_msg += (
+ f" The student's next exam in this course is {when}. Weight the"
+ " questions toward what an exam would actually test."
+ )
  _log_rag_uncovered(
  material,
  user_id=user_id,
@@ -1078,7 +1154,10 @@ def _absorb(quiz: Quiz, model: str) -> None:
  "quiz_agent produced no valid questions after wire-format validation"
  )
  # Never serve more than asked for (a generous top-up run can overshoot).
- return wire_questions[:num_questions]
+ return GeneratedQuiz(
+ questions=wire_questions[:num_questions],
+ exam_days_away=exam_days_away,
+ )
 
 @router.get("/config")
 def quiz_config():
@@ -1169,7 +1248,7 @@ async def generate_quiz(body: GenerateQuizBody, request: Request):
  # QUIZ_GENERATION_TIMEOUT_SEC (see _run) — cancelling the whole
  # coroutine here would discard a partial quiz the top-up handler
  # is designed to serve.
-questions = await _quiz_via_agent(
+generated = await _quiz_via_agent(
  user_id=body.user_id,
  course_id=course_id,
  concept_node_id=body.concept_node_id,
@@ -1180,6 +1259,8 @@ async def generate_quiz(body: GenerateQuizBody, request: Request):
  request_id=request_id,
  model_pref=body.model_pref,
  )
+ questions = generated.questions
+ exam_days_away = generated.exam_days_away
  except HTTPException:
  # The 404 for an unknown concept node is raised before the agent call;
  # never swallow a known HTTP state.
@@ -1224,13 +1305,21 @@ async def generate_quiz(body: GenerateQuizBody, request: Request):
  ) from e
 
  quiz_id = str(uuid.uuid4())
-table("quiz_attempts").insert({
+attempt_row = {
  "id": quiz_id,
  "user_id": body.user_id,
  "concept_node_id": body.concept_node_id,
  "difficulty": body.difficulty,
  "questions_json": encrypt_json(questions),
- })
+ }
+ # H3/#555: recorded so the question "do deadline-aware quizzes perform
+ # differently?" is answerable later. Omitted when unknown rather than
+ # written as an explicit null, mirroring apply_graph_update's rule: an
+ # environment that took this code before the migration keeps generating
+ # instead of 400ing on a column PostgREST's schema cache doesn't have.
+ if exam_days_away is not None:
+ attempt_row["exam_days_away"] = exam_days_away
+ _insert_attempt(attempt_row)
  # #117: quiz.started once the attempt row exists. num_questions is the
  # actual generated count (the agent may return fewer than requested).
  events_service.log_event(

backend/routes/study_guide.py

@@ -25,6 +25,7 @@
 )
 from services.graph_service import get_courses as graph_get_courses
 from services.auth_guard import require_self
+from services.exam_proximity import is_exam
 from services.encryption import (
  decrypt_if_present,
  decrypt_json,
@@ -303,15 +304,11 @@ def get_exams(
  order="due_date.asc",
  ) or []
 
- exam_keywords = ["exam", "midterm", "final", "quiz"]
- exams = []
- for a in all_assignments:
- atype = (a.get("assignment_type") or "").lower()
- title = (a.get("title") or "").lower()
- if atype == "exam" or any(kw in title for kw in exam_keywords):
- exams.append(a)
-
- return {"exams": exams}
+ # The heuristic moved to services/exam_proximity.py (#555), which needed
+ # the same question answered on the quiz path. One definition, two
+ # callers — a second copy would drift, which is the failure #557 spent a
+ # workstream undoing.
+ return {"exams": [a for a in all_assignments if is_exam(a)]}
 
 
 @router.get("/{user_id}/guide")

backend/scripts/benchmark_quiz.py

@@ -147,7 +147,10 @@ async def _generate_quiz_for_async(concept_name: str) -> list[dict]:
  so this exercises the real production grounding path rather than
  generating an ungrounded quiz.
  """
- return await _quiz_via_agent(
+ # `.questions`: _quiz_via_agent returns a GeneratedQuiz (#555) so the
+ # exam-proximity value it resolved can reach the attempt row. This bench
+ # only wants the questions.
+ generated = await _quiz_via_agent(
  user_id="quizfix-user-0001",
  course_id=FIXTURE_COURSE_ID,
  concept_node_id="quizfix-node-0001",
@@ -157,6 +160,7 @@ async def _generate_quiz_for_async(concept_name: str) -> list[dict]:
  use_shared_context=False,
  request_id="quizfix-bench",
  )
+ return generated.questions
 
 
 def generate_quiz_for(concept_name: str) -> list[dict]:

backend/db/migrations/20260822090747_quiz_attempts_exam_days_away.sql

@@ -0,0 +1,29 @@
+-- H3 (#555): how many days away the student's next exam was when this quiz
+-- was generated.
+--
+-- Stored on the attempt, not merely used in the prompt, because the point of
+-- the issue is to be able to ASK LATER whether deadline-aware quizzes perform
+-- differently. A value that only ever reached a prompt leaves no way to
+-- compare a quiz taken three days before a midterm against one taken in week
+-- two — the measurement is the deliverable, not the prompt line.
+--
+-- Nullable with no default, and the distinction is load-bearing:
+-- NULL = we did not know (no course, no enrollment, no dated exams, or the
+-- lookup failed — `days_until_next_exam` degrades to None rather
+-- than failing a generation)
+-- 0 = the exam is TODAY, which is the most actionable value the feature
+-- produces. A DEFAULT 0 would make every legacy row and every
+-- unknown look like exam day.
+--
+-- Dates only. This column is a day count derived from `assignments.due_date`,
+-- which is plaintext and indexed; no grade VALUE is read, stored or prompted
+-- anywhere on this path (points columns are encrypted per #521, and the audit
+-- flags that the current ToS does not clearly cover feeding grades to a
+-- model).
+--
+-- Additive and idempotent. Ordering still matters: PostgREST 400s on a column
+-- its schema cache doesn't have, and `generate_quiz` includes this key in the
+-- attempt INSERT — so this must be applied strictly before the code ships, as
+-- with 20260814051517.
+ALTER TABLE quiz_attempts
+ ADD COLUMN IF NOT EXISTS exam_days_away INTEGER;

backend/routes/quiz.py

@@ -42,6 +42,7 @@
 from services.encryption import encrypt_json, decrypt_json_column
 from services.graph_service import apply_graph_update
 from services.quiz_context_service import get_quiz_context, save_quiz_context
+from services.exam_proximity import PROMPT_HORIZON_DAYS, days_until_next_exam
 from services.quiz_distractors import build_distractor_profile
 from services.fingerprint import fingerprint
 from services.quiz_identity import question_hash, normalize_text
@@ -795,6 +796,50 @@ def _do_not_repeat_block(recent: list[RecentQuestion]) -> str:
  )
 
 
+def _insert_attempt(attempt_row: dict) -> None:
+ """Write the attempt, surviving a schema that predates `exam_days_away`.
+
+ The omit-when-None rule alone does NOT make a pre-migration environment
+ safe, and the failure is nastier than it looks: it strikes exactly the
+ students the feature is FOR (the ones with a dated upcoming exam), so it
+ presents as a random partial outage rather than an obvious missing
+ migration. And it strikes late — the agent has already run and been
+ billed — so an unhandled 400 here loses the generated quiz, writes no
+ attempt row, emits no `quiz.generation_failed`, and never refunds the
+ rate-limit slot.
+ #
+ Ordering is still the rule (migration before code, as with
+ 20260814051517). This is the seatbelt for the window where PostgREST has
+ not yet reloaded its schema cache, not a licence to deploy first.
+ """
+ try:
+ table("quiz_attempts").insert(attempt_row)
+ return
+ except Exception:
+ if "exam_days_away" not in attempt_row:
+ raise
+ retry = {k: v for k, v in attempt_row.items() if k != "exam_days_away"}
+ logger.warning(
+ "quiz: attempt insert failed with exam_days_away present; retrying "
+ "without it (is 20260822090747 applied?) quiz_id=%s",
+ attempt_row.get("id"),
+ )
+ table("quiz_attempts").insert(retry)
+
+
+class GeneratedQuiz(NamedTuple):
+ """What one generation produced.
+
+ `exam_days_away` rides back with the questions rather than being resolved
+ again by the caller: it is used for the prompt here and stored on the
+ attempt there, and two lookups could disagree if an exam were entered
+ between them.
+ """
+
+ questions: list[dict]
+ exam_days_away: int | None = None
+
+
 async def _quiz_via_agent(
  *,
  user_id: str,
@@ -806,7 +851,7 @@ async def _quiz_via_agent(
  use_shared_context: bool,
  request_id: str,
  model_pref: str | None = None,
-) -> list[dict]:
+) -> GeneratedQuiz:
  """Run quiz_agent and return questions in the legacy wire shape.
 
  The agent's tools (read_concepts_for_user, read_misconceptions_for_course)
@@ -849,42 +894,39 @@ async def _quiz_via_agent(
  difficulty_clause = (
  f"Generate {num_questions} {difficulty} questions for the student."
  )
- routing_msg = (
- f"{difficulty_clause} "
- 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:
- routing_msg += (
- " Also call read_misconceptions_for_course and use those misconceptions "
- "as distractors and probes."
- )
 
  # Course-material grounding does blocking network I/O (a Gemini
  # embedding call, bounded at 60s) plus sync Supabase reads. Run it in a
  # worker thread so a slow/stalled retrieval can't freeze this worker's
  # event loop for every other in-flight request. Matches the
  # asyncio.to_thread pattern used by the agent read tools.
  #
- # E6's recently-asked read is an independent Supabase read + decrypt, so
- # it runs CONCURRENTLY with grounding rather than after it — the two have
- # nothing to say to each other and serializing them would add the slower
- # one's latency to every generation.
+ # E6's recently-asked read and H3's exam-proximity lookup are independent
+ # Supabase reads, so all three run CONCURRENTLY rather than in sequence —
+ # they have nothing to say to each other, and serializing them would add
+ # every one of their latencies to every generation. Proximity costs
+ # several round-trips on its own, so running it before this block made it
+ # fully additive.
  #
- # return_exceptions=True because BOTH are best-effort context, and a bare
- # gather propagates the first failure straight out of generation: an
+ # return_exceptions=True because ALL THREE are best-effort context, and a
+ # bare gather propagates the first failure straight out of generation: an
  # unreadable past attempt would 502 a quiz that needed no history at all.
  # Each helper already degrades internally; this is the backstop for the
  # failure they cannot catch (an unexpected raise on the way in or out).
- material, recent = await asyncio.gather(
+ material, recent, exam_days_away = await asyncio.gather(
  asyncio.to_thread(_course_material, course_id, concept_name),
  asyncio.to_thread(
  recent_question_identities, user_id, concept_node_id
  ),
+ asyncio.to_thread(days_until_next_exam, user_id, course_id),
  return_exceptions=True,
  )
+ if isinstance(exam_days_away, BaseException):
+ logger.warning(
+ "quiz: exam-proximity lookup failed (%s); generating without it",
+ type(exam_days_away).__name__, exc_info=exam_days_away,
+ )
+ exam_days_away = None
  if isinstance(material, BaseException):
  logger.warning(
  "quiz: course-material assembly failed (%s); generating ungrounded",
@@ -900,6 +942,40 @@ async def _quiz_via_agent(
  "do-not-repeat list", type(recent).__name__, exc_info=recent,
  )
  recent = []
+
+ routing_msg = (
+ f"{difficulty_clause} "
+ 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:
+ routing_msg += (
+ " Also call read_misconceptions_for_course and use those misconceptions "
+ "as distractors and probes."
+ )
+ # H3/#555: one line, dates only. Says how near the deadline is and lets
+ # the model decide what that implies — not "make it harder", which would
+ # contradict the adaptive difficulty the student actually chose. Omitted
+ # entirely when unknown: "next exam: unknown" is prompt tokens spent to
+ # say nothing.
+ # Bounded by a horizon: a final dated 87 days out would otherwise put
+ # "there is an exam coming, weight toward what an exam tests" on EVERY
+ # quiz for the whole semester, which carries no proximity signal and
+ # steers week-two practice toward exam-style questions — the opposite of
+ # what this line is for. The stored `exam_days_away` is NOT clamped: the
+ # analytics want the real distance, including the far ones.
+ if exam_days_away is not None and exam_days_away <= PROMPT_HORIZON_DAYS:
+ when = (
+ "TODAY" if exam_days_away == 0
+ else "tomorrow" if exam_days_away == 1
+ else f"in {exam_days_away} days"
+ )
+ routing_msg += (
+ f" The student's next exam in this course is {when}. Weight the"
+ " questions toward what an exam would actually test."
+ )
  _log_rag_uncovered(
  material,
  user_id=user_id,
@@ -1078,7 +1154,10 @@ def _absorb(quiz: Quiz, model: str) -> None:
  "quiz_agent produced no valid questions after wire-format validation"
  )
  # Never serve more than asked for (a generous top-up run can overshoot).
- return wire_questions[:num_questions]
+ return GeneratedQuiz(
+ questions=wire_questions[:num_questions],
+ exam_days_away=exam_days_away,
+ )
 
 @router.get("/config")
 def quiz_config():
@@ -1169,7 +1248,7 @@ async def generate_quiz(body: GenerateQuizBody, request: Request):
  # QUIZ_GENERATION_TIMEOUT_SEC (see _run) — cancelling the whole
  # coroutine here would discard a partial quiz the top-up handler
  # is designed to serve.
-questions = await _quiz_via_agent(
+generated = await _quiz_via_agent(
  user_id=body.user_id,
  course_id=course_id,
  concept_node_id=body.concept_node_id,
@@ -1180,6 +1259,8 @@ async def generate_quiz(body: GenerateQuizBody, request: Request):
  request_id=request_id,
  model_pref=body.model_pref,
  )
+ questions = generated.questions
+ exam_days_away = generated.exam_days_away
  except HTTPException:
  # The 404 for an unknown concept node is raised before the agent call;
  # never swallow a known HTTP state.
@@ -1224,13 +1305,21 @@ async def generate_quiz(body: GenerateQuizBody, request: Request):
  ) from e
 
  quiz_id = str(uuid.uuid4())
-table("quiz_attempts").insert({
+attempt_row = {
  "id": quiz_id,
  "user_id": body.user_id,
  "concept_node_id": body.concept_node_id,
  "difficulty": body.difficulty,
  "questions_json": encrypt_json(questions),
- })
+ }
+ # H3/#555: recorded so the question "do deadline-aware quizzes perform
+ # differently?" is answerable later. Omitted when unknown rather than
+ # written as an explicit null, mirroring apply_graph_update's rule: an
+ # environment that took this code before the migration keeps generating
+ # instead of 400ing on a column PostgREST's schema cache doesn't have.
+ if exam_days_away is not None:
+ attempt_row["exam_days_away"] = exam_days_away
+ _insert_attempt(attempt_row)
  # #117: quiz.started once the attempt row exists. num_questions is the
  # actual generated count (the agent may return fewer than requested).
  events_service.log_event(

backend/routes/study_guide.py

@@ -25,6 +25,7 @@
 )
 from services.graph_service import get_courses as graph_get_courses
 from services.auth_guard import require_self
+from services.exam_proximity import is_exam
 from services.encryption import (
  decrypt_if_present,
  decrypt_json,
@@ -303,15 +304,11 @@ def get_exams(
  order="due_date.asc",
  ) or []
 
- exam_keywords = ["exam", "midterm", "final", "quiz"]
- exams = []
- for a in all_assignments:
- atype = (a.get("assignment_type") or "").lower()
- title = (a.get("title") or "").lower()
- if atype == "exam" or any(kw in title for kw in exam_keywords):
- exams.append(a)
-
- return {"exams": exams}
+ # The heuristic moved to services/exam_proximity.py (#555), which needed
+ # the same question answered on the quiz path. One definition, two
+ # callers — a second copy would drift, which is the failure #557 spent a
+ # workstream undoing.
+ return {"exams": [a for a in all_assignments if is_exam(a)]}
 
 
 @router.get("/{user_id}/guide")

backend/scripts/benchmark_quiz.py

@@ -147,7 +147,10 @@ async def _generate_quiz_for_async(concept_name: str) -> list[dict]:
  so this exercises the real production grounding path rather than
  generating an ungrounded quiz.
  """
- return await _quiz_via_agent(
+ # `.questions`: _quiz_via_agent returns a GeneratedQuiz (#555) so the
+ # exam-proximity value it resolved can reach the attempt row. This bench
+ # only wants the questions.
+ generated = await _quiz_via_agent(
  user_id="quizfix-user-0001",
  course_id=FIXTURE_COURSE_ID,
  concept_node_id="quizfix-node-0001",
@@ -157,6 +160,7 @@ async def _generate_quiz_for_async(concept_name: str) -> list[dict]:
  use_shared_context=False,
  request_id="quizfix-bench",
  )
+ return generated.questions
 
 
 def generate_quiz_for(concept_name: str) -> list[dict]:

backend/db/migrations/20260822090747_quiz_attempts_exam_days_away.sql

@@ -0,0 +1,29 @@
+-- H3 (#555): how many days away the student's next exam was when this quiz
+-- was generated.
+--
+-- Stored on the attempt, not merely used in the prompt, because the point of
+-- the issue is to be able to ASK LATER whether deadline-aware quizzes perform
+-- differently. A value that only ever reached a prompt leaves no way to
+-- compare a quiz taken three days before a midterm against one taken in week
+-- two — the measurement is the deliverable, not the prompt line.
+--
+-- Nullable with no default, and the distinction is load-bearing:
+-- NULL = we did not know (no course, no enrollment, no dated exams, or the
+-- lookup failed — `days_until_next_exam` degrades to None rather
+-- than failing a generation)
+-- 0 = the exam is TODAY, which is the most actionable value the feature
+-- produces. A DEFAULT 0 would make every legacy row and every
+-- unknown look like exam day.
+--
+-- Dates only. This column is a day count derived from `assignments.due_date`,
+-- which is plaintext and indexed; no grade VALUE is read, stored or prompted
+-- anywhere on this path (points columns are encrypted per #521, and the audit
+-- flags that the current ToS does not clearly cover feeding grades to a
+-- model).
+--
+-- Additive and idempotent. Ordering still matters: PostgREST 400s on a column
+-- its schema cache doesn't have, and `generate_quiz` includes this key in the
+-- attempt INSERT — so this must be applied strictly before the code ships, as
+-- with 20260814051517.
+ALTER TABLE quiz_attempts
+ ADD COLUMN IF NOT EXISTS exam_days_away INTEGER;

backend/routes/quiz.py

@@ -42,6 +42,7 @@
 from services.encryption import encrypt_json, decrypt_json_column
 from services.graph_service import apply_graph_update
 from services.quiz_context_service import get_quiz_context, save_quiz_context
+from services.exam_proximity import PROMPT_HORIZON_DAYS, days_until_next_exam
 from services.quiz_distractors import build_distractor_profile
 from services.fingerprint import fingerprint
 from services.quiz_identity import question_hash, normalize_text
@@ -795,6 +796,50 @@ def _do_not_repeat_block(recent: list[RecentQuestion]) -> str:
  )
 
 
+def _insert_attempt(attempt_row: dict) -> None:
+ """Write the attempt, surviving a schema that predates `exam_days_away`.
+
+ The omit-when-None rule alone does NOT make a pre-migration environment
+ safe, and the failure is nastier than it looks: it strikes exactly the
+ students the feature is FOR (the ones with a dated upcoming exam), so it
+ presents as a random partial outage rather than an obvious missing
+ migration. And it strikes late — the agent has already run and been
+ billed — so an unhandled 400 here loses the generated quiz, writes no
+ attempt row, emits no `quiz.generation_failed`, and never refunds the
+ rate-limit slot.
+ #
+ Ordering is still the rule (migration before code, as with
+ 20260814051517). This is the seatbelt for the window where PostgREST has
+ not yet reloaded its schema cache, not a licence to deploy first.
+ """
+ try:
+ table("quiz_attempts").insert(attempt_row)
+ return
+ except Exception:
+ if "exam_days_away" not in attempt_row:
+ raise
+ retry = {k: v for k, v in attempt_row.items() if k != "exam_days_away"}
+ logger.warning(
+ "quiz: attempt insert failed with exam_days_away present; retrying "
+ "without it (is 20260822090747 applied?) quiz_id=%s",
+ attempt_row.get("id"),
+ )
+ table("quiz_attempts").insert(retry)
+
+
+class GeneratedQuiz(NamedTuple):
+ """What one generation produced.
+
+ `exam_days_away` rides back with the questions rather than being resolved
+ again by the caller: it is used for the prompt here and stored on the
+ attempt there, and two lookups could disagree if an exam were entered
+ between them.
+ """
+
+ questions: list[dict]
+ exam_days_away: int | None = None
+
+
 async def _quiz_via_agent(
  *,
  user_id: str,
@@ -806,7 +851,7 @@ async def _quiz_via_agent(
  use_shared_context: bool,
  request_id: str,
  model_pref: str | None = None,
-) -> list[dict]:
+) -> GeneratedQuiz:
  """Run quiz_agent and return questions in the legacy wire shape.
 
  The agent's tools (read_concepts_for_user, read_misconceptions_for_course)
@@ -849,42 +894,39 @@ async def _quiz_via_agent(
  difficulty_clause = (
  f"Generate {num_questions} {difficulty} questions for the student."
  )
- routing_msg = (
- f"{difficulty_clause} "
- 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:
- routing_msg += (
- " Also call read_misconceptions_for_course and use those misconceptions "
- "as distractors and probes."
- )
 
  # Course-material grounding does blocking network I/O (a Gemini
  # embedding call, bounded at 60s) plus sync Supabase reads. Run it in a
  # worker thread so a slow/stalled retrieval can't freeze this worker's
  # event loop for every other in-flight request. Matches the
  # asyncio.to_thread pattern used by the agent read tools.
  #
- # E6's recently-asked read is an independent Supabase read + decrypt, so
- # it runs CONCURRENTLY with grounding rather than after it — the two have
- # nothing to say to each other and serializing them would add the slower
- # one's latency to every generation.
+ # E6's recently-asked read and H3's exam-proximity lookup are independent
+ # Supabase reads, so all three run CONCURRENTLY rather than in sequence —
+ # they have nothing to say to each other, and serializing them would add
+ # every one of their latencies to every generation. Proximity costs
+ # several round-trips on its own, so running it before this block made it
+ # fully additive.
  #
- # return_exceptions=True because BOTH are best-effort context, and a bare
- # gather propagates the first failure straight out of generation: an
+ # return_exceptions=True because ALL THREE are best-effort context, and a
+ # bare gather propagates the first failure straight out of generation: an
  # unreadable past attempt would 502 a quiz that needed no history at all.
  # Each helper already degrades internally; this is the backstop for the
  # failure they cannot catch (an unexpected raise on the way in or out).
- material, recent = await asyncio.gather(
+ material, recent, exam_days_away = await asyncio.gather(
  asyncio.to_thread(_course_material, course_id, concept_name),
  asyncio.to_thread(
  recent_question_identities, user_id, concept_node_id
  ),
+ asyncio.to_thread(days_until_next_exam, user_id, course_id),
  return_exceptions=True,
  )
+ if isinstance(exam_days_away, BaseException):
+ logger.warning(
+ "quiz: exam-proximity lookup failed (%s); generating without it",
+ type(exam_days_away).__name__, exc_info=exam_days_away,
+ )
+ exam_days_away = None
  if isinstance(material, BaseException):
  logger.warning(
  "quiz: course-material assembly failed (%s); generating ungrounded",
@@ -900,6 +942,40 @@ async def _quiz_via_agent(
  "do-not-repeat list", type(recent).__name__, exc_info=recent,
  )
  recent = []
+
+ routing_msg = (
+ f"{difficulty_clause} "
+ 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:
+ routing_msg += (
+ " Also call read_misconceptions_for_course and use those misconceptions "
+ "as distractors and probes."
+ )
+ # H3/#555: one line, dates only. Says how near the deadline is and lets
+ # the model decide what that implies — not "make it harder", which would
+ # contradict the adaptive difficulty the student actually chose. Omitted
+ # entirely when unknown: "next exam: unknown" is prompt tokens spent to
+ # say nothing.
+ # Bounded by a horizon: a final dated 87 days out would otherwise put
+ # "there is an exam coming, weight toward what an exam tests" on EVERY
+ # quiz for the whole semester, which carries no proximity signal and
+ # steers week-two practice toward exam-style questions — the opposite of
+ # what this line is for. The stored `exam_days_away` is NOT clamped: the
+ # analytics want the real distance, including the far ones.
+ if exam_days_away is not None and exam_days_away <= PROMPT_HORIZON_DAYS:
+ when = (
+ "TODAY" if exam_days_away == 0
+ else "tomorrow" if exam_days_away == 1
+ else f"in {exam_days_away} days"
+ )
+ routing_msg += (
+ f" The student's next exam in this course is {when}. Weight the"
+ " questions toward what an exam would actually test."
+ )
  _log_rag_uncovered(
  material,
  user_id=user_id,
@@ -1078,7 +1154,10 @@ def _absorb(quiz: Quiz, model: str) -> None:
  "quiz_agent produced no valid questions after wire-format validation"
  )
  # Never serve more than asked for (a generous top-up run can overshoot).
- return wire_questions[:num_questions]
+ return GeneratedQuiz(
+ questions=wire_questions[:num_questions],
+ exam_days_away=exam_days_away,
+ )
 
 @router.get("/config")
 def quiz_config():
@@ -1169,7 +1248,7 @@ async def generate_quiz(body: GenerateQuizBody, request: Request):
  # QUIZ_GENERATION_TIMEOUT_SEC (see _run) — cancelling the whole
  # coroutine here would discard a partial quiz the top-up handler
  # is designed to serve.
-questions = await _quiz_via_agent(
+generated = await _quiz_via_agent(
  user_id=body.user_id,
  course_id=course_id,
  concept_node_id=body.concept_node_id,
@@ -1180,6 +1259,8 @@ async def generate_quiz(body: GenerateQuizBody, request: Request):
  request_id=request_id,
  model_pref=body.model_pref,
  )
+ questions = generated.questions
+ exam_days_away = generated.exam_days_away
  except HTTPException:
  # The 404 for an unknown concept node is raised before the agent call;
  # never swallow a known HTTP state.
@@ -1224,13 +1305,21 @@ async def generate_quiz(body: GenerateQuizBody, request: Request):
  ) from e
 
  quiz_id = str(uuid.uuid4())
-table("quiz_attempts").insert({
+attempt_row = {
  "id": quiz_id,
  "user_id": body.user_id,
  "concept_node_id": body.concept_node_id,
  "difficulty": body.difficulty,
  "questions_json": encrypt_json(questions),
- })
+ }
+ # H3/#555: recorded so the question "do deadline-aware quizzes perform
+ # differently?" is answerable later. Omitted when unknown rather than
+ # written as an explicit null, mirroring apply_graph_update's rule: an
+ # environment that took this code before the migration keeps generating
+ # instead of 400ing on a column PostgREST's schema cache doesn't have.
+ if exam_days_away is not None:
+ attempt_row["exam_days_away"] = exam_days_away
+ _insert_attempt(attempt_row)
  # #117: quiz.started once the attempt row exists. num_questions is the
  # actual generated count (the agent may return fewer than requested).
  events_service.log_event(

backend/routes/study_guide.py

@@ -25,6 +25,7 @@
 )
 from services.graph_service import get_courses as graph_get_courses
 from services.auth_guard import require_self
+from services.exam_proximity import is_exam
 from services.encryption import (
  decrypt_if_present,
  decrypt_json,
@@ -303,15 +304,11 @@ def get_exams(
  order="due_date.asc",
  ) or []
 
- exam_keywords = ["exam", "midterm", "final", "quiz"]
- exams = []
- for a in all_assignments:
- atype = (a.get("assignment_type") or "").lower()
- title = (a.get("title") or "").lower()
- if atype == "exam" or any(kw in title for kw in exam_keywords):
- exams.append(a)
-
- return {"exams": exams}
+ # The heuristic moved to services/exam_proximity.py (#555), which needed
+ # the same question answered on the quiz path. One definition, two
+ # callers — a second copy would drift, which is the failure #557 spent a
+ # workstream undoing.
+ return {"exams": [a for a in all_assignments if is_exam(a)]}
 
 
 @router.get("/{user_id}/guide")

backend/scripts/benchmark_quiz.py

@@ -147,7 +147,10 @@ async def _generate_quiz_for_async(concept_name: str) -> list[dict]:
  so this exercises the real production grounding path rather than
  generating an ungrounded quiz.
  """
- return await _quiz_via_agent(
+ # `.questions`: _quiz_via_agent returns a GeneratedQuiz (#555) so the
+ # exam-proximity value it resolved can reach the attempt row. This bench
+ # only wants the questions.
+ generated = await _quiz_via_agent(
  user_id="quizfix-user-0001",
  course_id=FIXTURE_COURSE_ID,
  concept_node_id="quizfix-node-0001",
@@ -157,6 +160,7 @@ async def _generate_quiz_for_async(concept_name: str) -> list[dict]:
  use_shared_context=False,
  request_id="quizfix-bench",
  )
+ return generated.questions
 
 
 def generate_quiz_for(concept_name: str) -> list[dict]:

backend/db/migrations/20260822090747_quiz_attempts_exam_days_away.sql

@@ -0,0 +1,29 @@
+-- H3 (#555): how many days away the student's next exam was when this quiz
+-- was generated.
+--
+-- Stored on the attempt, not merely used in the prompt, because the point of
+-- the issue is to be able to ASK LATER whether deadline-aware quizzes perform
+-- differently. A value that only ever reached a prompt leaves no way to
+-- compare a quiz taken three days before a midterm against one taken in week
+-- two — the measurement is the deliverable, not the prompt line.
+--
+-- Nullable with no default, and the distinction is load-bearing:
+-- NULL = we did not know (no course, no enrollment, no dated exams, or the
+-- lookup failed — `days_until_next_exam` degrades to None rather
+-- than failing a generation)
+-- 0 = the exam is TODAY, which is the most actionable value the feature
+-- produces. A DEFAULT 0 would make every legacy row and every
+-- unknown look like exam day.
+--
+-- Dates only. This column is a day count derived from `assignments.due_date`,
+-- which is plaintext and indexed; no grade VALUE is read, stored or prompted
+-- anywhere on this path (points columns are encrypted per #521, and the audit
+-- flags that the current ToS does not clearly cover feeding grades to a
+-- model).
+--
+-- Additive and idempotent. Ordering still matters: PostgREST 400s on a column
+-- its schema cache doesn't have, and `generate_quiz` includes this key in the
+-- attempt INSERT — so this must be applied strictly before the code ships, as
+-- with 20260814051517.
+ALTER TABLE quiz_attempts
+ ADD COLUMN IF NOT EXISTS exam_days_away INTEGER;

backend/routes/quiz.py

@@ -42,6 +42,7 @@
 from services.encryption import encrypt_json, decrypt_json_column
 from services.graph_service import apply_graph_update
 from services.quiz_context_service import get_quiz_context, save_quiz_context
+from services.exam_proximity import PROMPT_HORIZON_DAYS, days_until_next_exam
 from services.quiz_distractors import build_distractor_profile
 from services.fingerprint import fingerprint
 from services.quiz_identity import question_hash, normalize_text
@@ -795,6 +796,50 @@ def _do_not_repeat_block(recent: list[RecentQuestion]) -> str:
  )
 
 
+def _insert_attempt(attempt_row: dict) -> None:
+ """Write the attempt, surviving a schema that predates `exam_days_away`.
+
+ The omit-when-None rule alone does NOT make a pre-migration environment
+ safe, and the failure is nastier than it looks: it strikes exactly the
+ students the feature is FOR (the ones with a dated upcoming exam), so it
+ presents as a random partial outage rather than an obvious missing
+ migration. And it strikes late — the agent has already run and been
+ billed — so an unhandled 400 here loses the generated quiz, writes no
+ attempt row, emits no `quiz.generation_failed`, and never refunds the
+ rate-limit slot.
+ #
+ Ordering is still the rule (migration before code, as with
+ 20260814051517). This is the seatbelt for the window where PostgREST has
+ not yet reloaded its schema cache, not a licence to deploy first.
+ """
+ try:
+ table("quiz_attempts").insert(attempt_row)
+ return
+ except Exception:
+ if "exam_days_away" not in attempt_row:
+ raise
+ retry = {k: v for k, v in attempt_row.items() if k != "exam_days_away"}
+ logger.warning(
+ "quiz: attempt insert failed with exam_days_away present; retrying "
+ "without it (is 20260822090747 applied?) quiz_id=%s",
+ attempt_row.get("id"),
+ )
+ table("quiz_attempts").insert(retry)
+
+
+class GeneratedQuiz(NamedTuple):
+ """What one generation produced.
+
+ `exam_days_away` rides back with the questions rather than being resolved
+ again by the caller: it is used for the prompt here and stored on the
+ attempt there, and two lookups could disagree if an exam were entered
+ between them.
+ """
+
+ questions: list[dict]
+ exam_days_away: int | None = None
+
+
 async def _quiz_via_agent(
  *,
  user_id: str,
@@ -806,7 +851,7 @@ async def _quiz_via_agent(
  use_shared_context: bool,
  request_id: str,
  model_pref: str | None = None,
-) -> list[dict]:
+) -> GeneratedQuiz:
  """Run quiz_agent and return questions in the legacy wire shape.
 
  The agent's tools (read_concepts_for_user, read_misconceptions_for_course)
@@ -849,42 +894,39 @@ async def _quiz_via_agent(
  difficulty_clause = (
  f"Generate {num_questions} {difficulty} questions for the student."
  )
- routing_msg = (
- f"{difficulty_clause} "
- 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:
- routing_msg += (
- " Also call read_misconceptions_for_course and use those misconceptions "
- "as distractors and probes."
- )
 
  # Course-material grounding does blocking network I/O (a Gemini
  # embedding call, bounded at 60s) plus sync Supabase reads. Run it in a
  # worker thread so a slow/stalled retrieval can't freeze this worker's
  # event loop for every other in-flight request. Matches the
  # asyncio.to_thread pattern used by the agent read tools.
  #
- # E6's recently-asked read is an independent Supabase read + decrypt, so
- # it runs CONCURRENTLY with grounding rather than after it — the two have
- # nothing to say to each other and serializing them would add the slower
- # one's latency to every generation.
+ # E6's recently-asked read and H3's exam-proximity lookup are independent
+ # Supabase reads, so all three run CONCURRENTLY rather than in sequence —
+ # they have nothing to say to each other, and serializing them would add
+ # every one of their latencies to every generation. Proximity costs
+ # several round-trips on its own, so running it before this block made it
+ # fully additive.
  #
- # return_exceptions=True because BOTH are best-effort context, and a bare
- # gather propagates the first failure straight out of generation: an
+ # return_exceptions=True because ALL THREE are best-effort context, and a
+ # bare gather propagates the first failure straight out of generation: an
  # unreadable past attempt would 502 a quiz that needed no history at all.
  # Each helper already degrades internally; this is the backstop for the
  # failure they cannot catch (an unexpected raise on the way in or out).
- material, recent = await asyncio.gather(
+ material, recent, exam_days_away = await asyncio.gather(
  asyncio.to_thread(_course_material, course_id, concept_name),
  asyncio.to_thread(
  recent_question_identities, user_id, concept_node_id
  ),
+ asyncio.to_thread(days_until_next_exam, user_id, course_id),
  return_exceptions=True,
  )
+ if isinstance(exam_days_away, BaseException):
+ logger.warning(
+ "quiz: exam-proximity lookup failed (%s); generating without it",
+ type(exam_days_away).__name__, exc_info=exam_days_away,
+ )
+ exam_days_away = None
  if isinstance(material, BaseException):
  logger.warning(
  "quiz: course-material assembly failed (%s); generating ungrounded",
@@ -900,6 +942,40 @@ async def _quiz_via_agent(
  "do-not-repeat list", type(recent).__name__, exc_info=recent,
  )
  recent = []
+
+ routing_msg = (
+ f"{difficulty_clause} "
+ 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:
+ routing_msg += (
+ " Also call read_misconceptions_for_course and use those misconceptions "
+ "as distractors and probes."
+ )
+ # H3/#555: one line, dates only. Says how near the deadline is and lets
+ # the model decide what that implies — not "make it harder", which would
+ # contradict the adaptive difficulty the student actually chose. Omitted
+ # entirely when unknown: "next exam: unknown" is prompt tokens spent to
+ # say nothing.
+ # Bounded by a horizon: a final dated 87 days out would otherwise put
+ # "there is an exam coming, weight toward what an exam tests" on EVERY
+ # quiz for the whole semester, which carries no proximity signal and
+ # steers week-two practice toward exam-style questions — the opposite of
+ # what this line is for. The stored `exam_days_away` is NOT clamped: the
+ # analytics want the real distance, including the far ones.
+ if exam_days_away is not None and exam_days_away <= PROMPT_HORIZON_DAYS:
+ when = (
+ "TODAY" if exam_days_away == 0
+ else "tomorrow" if exam_days_away == 1
+ else f"in {exam_days_away} days"
+ )
+ routing_msg += (
+ f" The student's next exam in this course is {when}. Weight the"
+ " questions toward what an exam would actually test."
+ )
  _log_rag_uncovered(
  material,
  user_id=user_id,
@@ -1078,7 +1154,10 @@ def _absorb(quiz: Quiz, model: str) -> None:
  "quiz_agent produced no valid questions after wire-format validation"
  )
  # Never serve more than asked for (a generous top-up run can overshoot).
- return wire_questions[:num_questions]
+ return GeneratedQuiz(
+ questions=wire_questions[:num_questions],
+ exam_days_away=exam_days_away,
+ )
 
 @router.get("/config")
 def quiz_config():
@@ -1169,7 +1248,7 @@ async def generate_quiz(body: GenerateQuizBody, request: Request):
  # QUIZ_GENERATION_TIMEOUT_SEC (see _run) — cancelling the whole
  # coroutine here would discard a partial quiz the top-up handler
  # is designed to serve.
-questions = await _quiz_via_agent(
+generated = await _quiz_via_agent(
  user_id=body.user_id,
  course_id=course_id,
  concept_node_id=body.concept_node_id,
@@ -1180,6 +1259,8 @@ async def generate_quiz(body: GenerateQuizBody, request: Request):
  request_id=request_id,
  model_pref=body.model_pref,
  )
+ questions = generated.questions
+ exam_days_away = generated.exam_days_away
  except HTTPException:
  # The 404 for an unknown concept node is raised before the agent call;
  # never swallow a known HTTP state.
@@ -1224,13 +1305,21 @@ async def generate_quiz(body: GenerateQuizBody, request: Request):
  ) from e
 
  quiz_id = str(uuid.uuid4())
-table("quiz_attempts").insert({
+attempt_row = {
  "id": quiz_id,
  "user_id": body.user_id,
  "concept_node_id": body.concept_node_id,
  "difficulty": body.difficulty,
  "questions_json": encrypt_json(questions),
- })
+ }
+ # H3/#555: recorded so the question "do deadline-aware quizzes perform
+ # differently?" is answerable later. Omitted when unknown rather than
+ # written as an explicit null, mirroring apply_graph_update's rule: an
+ # environment that took this code before the migration keeps generating
+ # instead of 400ing on a column PostgREST's schema cache doesn't have.
+ if exam_days_away is not None:
+ attempt_row["exam_days_away"] = exam_days_away
+ _insert_attempt(attempt_row)
  # #117: quiz.started once the attempt row exists. num_questions is the
  # actual generated count (the agent may return fewer than requested).
  events_service.log_event(

backend/routes/study_guide.py

@@ -25,6 +25,7 @@
 )
 from services.graph_service import get_courses as graph_get_courses
 from services.auth_guard import require_self
+from services.exam_proximity import is_exam
 from services.encryption import (
  decrypt_if_present,
  decrypt_json,
@@ -303,15 +304,11 @@ def get_exams(
  order="due_date.asc",
  ) or []
 
- exam_keywords = ["exam", "midterm", "final", "quiz"]
- exams = []
- for a in all_assignments:
- atype = (a.get("assignment_type") or "").lower()
- title = (a.get("title") or "").lower()
- if atype == "exam" or any(kw in title for kw in exam_keywords):
- exams.append(a)
-
- return {"exams": exams}
+ # The heuristic moved to services/exam_proximity.py (#555), which needed
+ # the same question answered on the quiz path. One definition, two
+ # callers — a second copy would drift, which is the failure #557 spent a
+ # workstream undoing.
+ return {"exams": [a for a in all_assignments if is_exam(a)]}
 
 
 @router.get("/{user_id}/guide")

backend/scripts/benchmark_quiz.py

@@ -147,7 +147,10 @@ async def _generate_quiz_for_async(concept_name: str) -> list[dict]:
  so this exercises the real production grounding path rather than
  generating an ungrounded quiz.
  """
- return await _quiz_via_agent(
+ # `.questions`: _quiz_via_agent returns a GeneratedQuiz (#555) so the
+ # exam-proximity value it resolved can reach the attempt row. This bench
+ # only wants the questions.
+ generated = await _quiz_via_agent(
  user_id="quizfix-user-0001",
  course_id=FIXTURE_COURSE_ID,
  concept_node_id="quizfix-node-0001",
@@ -157,6 +160,7 @@ async def _generate_quiz_for_async(concept_name: str) -> list[dict]:
  use_shared_context=False,
  request_id="quizfix-bench",
  )
+ return generated.questions
 
 
 def generate_quiz_for(concept_name: str) -> list[dict]:

backend/db/migrations/20260822090747_quiz_attempts_exam_days_away.sql

@@ -0,0 +1,29 @@
+-- H3 (#555): how many days away the student's next exam was when this quiz
+-- was generated.
+--
+-- Stored on the attempt, not merely used in the prompt, because the point of
+-- the issue is to be able to ASK LATER whether deadline-aware quizzes perform
+-- differently. A value that only ever reached a prompt leaves no way to
+-- compare a quiz taken three days before a midterm against one taken in week
+-- two — the measurement is the deliverable, not the prompt line.
+--
+-- Nullable with no default, and the distinction is load-bearing:
+-- NULL = we did not know (no course, no enrollment, no dated exams, or the
+-- lookup failed — `days_until_next_exam` degrades to None rather
+-- than failing a generation)
+-- 0 = the exam is TODAY, which is the most actionable value the feature
+-- produces. A DEFAULT 0 would make every legacy row and every
+-- unknown look like exam day.
+--
+-- Dates only. This column is a day count derived from `assignments.due_date`,
+-- which is plaintext and indexed; no grade VALUE is read, stored or prompted
+-- anywhere on this path (points columns are encrypted per #521, and the audit
+-- flags that the current ToS does not clearly cover feeding grades to a
+-- model).
+--
+-- Additive and idempotent. Ordering still matters: PostgREST 400s on a column
+-- its schema cache doesn't have, and `generate_quiz` includes this key in the
+-- attempt INSERT — so this must be applied strictly before the code ships, as
+-- with 20260814051517.
+ALTER TABLE quiz_attempts
+ ADD COLUMN IF NOT EXISTS exam_days_away INTEGER;

backend/routes/quiz.py

@@ -42,6 +42,7 @@
 from services.encryption import encrypt_json, decrypt_json_column
 from services.graph_service import apply_graph_update
 from services.quiz_context_service import get_quiz_context, save_quiz_context
+from services.exam_proximity import PROMPT_HORIZON_DAYS, days_until_next_exam
 from services.quiz_distractors import build_distractor_profile
 from services.fingerprint import fingerprint
 from services.quiz_identity import question_hash, normalize_text
@@ -795,6 +796,50 @@ def _do_not_repeat_block(recent: list[RecentQuestion]) -> str:
  )
 
 
+def _insert_attempt(attempt_row: dict) -> None:
+ """Write the attempt, surviving a schema that predates `exam_days_away`.
+
+ The omit-when-None rule alone does NOT make a pre-migration environment
+ safe, and the failure is nastier than it looks: it strikes exactly the
+ students the feature is FOR (the ones with a dated upcoming exam), so it
+ presents as a random partial outage rather than an obvious missing
+ migration. And it strikes late — the agent has already run and been
+ billed — so an unhandled 400 here loses the generated quiz, writes no
+ attempt row, emits no `quiz.generation_failed`, and never refunds the
+ rate-limit slot.
+ #
+ Ordering is still the rule (migration before code, as with
+ 20260814051517). This is the seatbelt for the window where PostgREST has
+ not yet reloaded its schema cache, not a licence to deploy first.
+ """
+ try:
+ table("quiz_attempts").insert(attempt_row)
+ return
+ except Exception:
+ if "exam_days_away" not in attempt_row:
+ raise
+ retry = {k: v for k, v in attempt_row.items() if k != "exam_days_away"}
+ logger.warning(
+ "quiz: attempt insert failed with exam_days_away present; retrying "
+ "without it (is 20260822090747 applied?) quiz_id=%s",
+ attempt_row.get("id"),
+ )
+ table("quiz_attempts").insert(retry)
+
+
+class GeneratedQuiz(NamedTuple):
+ """What one generation produced.
+
+ `exam_days_away` rides back with the questions rather than being resolved
+ again by the caller: it is used for the prompt here and stored on the
+ attempt there, and two lookups could disagree if an exam were entered
+ between them.
+ """
+
+ questions: list[dict]
+ exam_days_away: int | None = None
+
+
 async def _quiz_via_agent(
  *,
  user_id: str,
@@ -806,7 +851,7 @@ async def _quiz_via_agent(
  use_shared_context: bool,
  request_id: str,
  model_pref: str | None = None,
-) -> list[dict]:
+) -> GeneratedQuiz:
  """Run quiz_agent and return questions in the legacy wire shape.
 
  The agent's tools (read_concepts_for_user, read_misconceptions_for_course)
@@ -849,42 +894,39 @@ async def _quiz_via_agent(
  difficulty_clause = (
  f"Generate {num_questions} {difficulty} questions for the student."
  )
- routing_msg = (
- f"{difficulty_clause} "
- 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:
- routing_msg += (
- " Also call read_misconceptions_for_course and use those misconceptions "
- "as distractors and probes."
- )
 
  # Course-material grounding does blocking network I/O (a Gemini
  # embedding call, bounded at 60s) plus sync Supabase reads. Run it in a
  # worker thread so a slow/stalled retrieval can't freeze this worker's
  # event loop for every other in-flight request. Matches the
  # asyncio.to_thread pattern used by the agent read tools.
  #
- # E6's recently-asked read is an independent Supabase read + decrypt, so
- # it runs CONCURRENTLY with grounding rather than after it — the two have
- # nothing to say to each other and serializing them would add the slower
- # one's latency to every generation.
+ # E6's recently-asked read and H3's exam-proximity lookup are independent
+ # Supabase reads, so all three run CONCURRENTLY rather than in sequence —
+ # they have nothing to say to each other, and serializing them would add
+ # every one of their latencies to every generation. Proximity costs
+ # several round-trips on its own, so running it before this block made it
+ # fully additive.
  #
- # return_exceptions=True because BOTH are best-effort context, and a bare
- # gather propagates the first failure straight out of generation: an
+ # return_exceptions=True because ALL THREE are best-effort context, and a
+ # bare gather propagates the first failure straight out of generation: an
  # unreadable past attempt would 502 a quiz that needed no history at all.
  # Each helper already degrades internally; this is the backstop for the
  # failure they cannot catch (an unexpected raise on the way in or out).
- material, recent = await asyncio.gather(
+ material, recent, exam_days_away = await asyncio.gather(
  asyncio.to_thread(_course_material, course_id, concept_name),
  asyncio.to_thread(
  recent_question_identities, user_id, concept_node_id
  ),
+ asyncio.to_thread(days_until_next_exam, user_id, course_id),
  return_exceptions=True,
  )
+ if isinstance(exam_days_away, BaseException):
+ logger.warning(
+ "quiz: exam-proximity lookup failed (%s); generating without it",
+ type(exam_days_away).__name__, exc_info=exam_days_away,
+ )
+ exam_days_away = None
  if isinstance(material, BaseException):
  logger.warning(
  "quiz: course-material assembly failed (%s); generating ungrounded",
@@ -900,6 +942,40 @@ async def _quiz_via_agent(
  "do-not-repeat list", type(recent).__name__, exc_info=recent,
  )
  recent = []
+
+ routing_msg = (
+ f"{difficulty_clause} "
+ 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:
+ routing_msg += (
+ " Also call read_misconceptions_for_course and use those misconceptions "
+ "as distractors and probes."
+ )
+ # H3/#555: one line, dates only. Says how near the deadline is and lets
+ # the model decide what that implies — not "make it harder", which would
+ # contradict the adaptive difficulty the student actually chose. Omitted
+ # entirely when unknown: "next exam: unknown" is prompt tokens spent to
+ # say nothing.
+ # Bounded by a horizon: a final dated 87 days out would otherwise put
+ # "there is an exam coming, weight toward what an exam tests" on EVERY
+ # quiz for the whole semester, which carries no proximity signal and
+ # steers week-two practice toward exam-style questions — the opposite of
+ # what this line is for. The stored `exam_days_away` is NOT clamped: the
+ # analytics want the real distance, including the far ones.
+ if exam_days_away is not None and exam_days_away <= PROMPT_HORIZON_DAYS:
+ when = (
+ "TODAY" if exam_days_away == 0
+ else "tomorrow" if exam_days_away == 1
+ else f"in {exam_days_away} days"
+ )
+ routing_msg += (
+ f" The student's next exam in this course is {when}. Weight the"
+ " questions toward what an exam would actually test."
+ )
  _log_rag_uncovered(
  material,
  user_id=user_id,
@@ -1078,7 +1154,10 @@ def _absorb(quiz: Quiz, model: str) -> None:
  "quiz_agent produced no valid questions after wire-format validation"
  )
  # Never serve more than asked for (a generous top-up run can overshoot).
- return wire_questions[:num_questions]
+ return GeneratedQuiz(
+ questions=wire_questions[:num_questions],
+ exam_days_away=exam_days_away,
+ )
 
 @router.get("/config")
 def quiz_config():
@@ -1169,7 +1248,7 @@ async def generate_quiz(body: GenerateQuizBody, request: Request):
  # QUIZ_GENERATION_TIMEOUT_SEC (see _run) — cancelling the whole
  # coroutine here would discard a partial quiz the top-up handler
  # is designed to serve.
-questions = await _quiz_via_agent(
+generated = await _quiz_via_agent(
  user_id=body.user_id,
  course_id=course_id,
  concept_node_id=body.concept_node_id,
@@ -1180,6 +1259,8 @@ async def generate_quiz(body: GenerateQuizBody, request: Request):
  request_id=request_id,
  model_pref=body.model_pref,
  )
+ questions = generated.questions
+ exam_days_away = generated.exam_days_away
  except HTTPException:
  # The 404 for an unknown concept node is raised before the agent call;
  # never swallow a known HTTP state.
@@ -1224,13 +1305,21 @@ async def generate_quiz(body: GenerateQuizBody, request: Request):
  ) from e
 
  quiz_id = str(uuid.uuid4())
-table("quiz_attempts").insert({
+attempt_row = {
  "id": quiz_id,
  "user_id": body.user_id,
  "concept_node_id": body.concept_node_id,
  "difficulty": body.difficulty,
  "questions_json": encrypt_json(questions),
- })
+ }
+ # H3/#555: recorded so the question "do deadline-aware quizzes perform
+ # differently?" is answerable later. Omitted when unknown rather than
+ # written as an explicit null, mirroring apply_graph_update's rule: an
+ # environment that took this code before the migration keeps generating
+ # instead of 400ing on a column PostgREST's schema cache doesn't have.
+ if exam_days_away is not None:
+ attempt_row["exam_days_away"] = exam_days_away
+ _insert_attempt(attempt_row)
  # #117: quiz.started once the attempt row exists. num_questions is the
  # actual generated count (the agent may return fewer than requested).
  events_service.log_event(

backend/routes/study_guide.py

@@ -25,6 +25,7 @@
 )
 from services.graph_service import get_courses as graph_get_courses
 from services.auth_guard import require_self
+from services.exam_proximity import is_exam
 from services.encryption import (
  decrypt_if_present,
  decrypt_json,
@@ -303,15 +304,11 @@ def get_exams(
  order="due_date.asc",
  ) or []
 
- exam_keywords = ["exam", "midterm", "final", "quiz"]
- exams = []
- for a in all_assignments:
- atype = (a.get("assignment_type") or "").lower()
- title = (a.get("title") or "").lower()
- if atype == "exam" or any(kw in title for kw in exam_keywords):
- exams.append(a)
-
- return {"exams": exams}
+ # The heuristic moved to services/exam_proximity.py (#555), which needed
+ # the same question answered on the quiz path. One definition, two
+ # callers — a second copy would drift, which is the failure #557 spent a
+ # workstream undoing.
+ return {"exams": [a for a in all_assignments if is_exam(a)]}
 
 
 @router.get("/{user_id}/guide")

backend/scripts/benchmark_quiz.py

@@ -147,7 +147,10 @@ async def _generate_quiz_for_async(concept_name: str) -> list[dict]:
  so this exercises the real production grounding path rather than
  generating an ungrounded quiz.
  """
- return await _quiz_via_agent(
+ # `.questions`: _quiz_via_agent returns a GeneratedQuiz (#555) so the
+ # exam-proximity value it resolved can reach the attempt row. This bench
+ # only wants the questions.
+ generated = await _quiz_via_agent(
  user_id="quizfix-user-0001",
  course_id=FIXTURE_COURSE_ID,
  concept_node_id="quizfix-node-0001",
@@ -157,6 +160,7 @@ async def _generate_quiz_for_async(concept_name: str) -> list[dict]:
  use_shared_context=False,
  request_id="quizfix-bench",
  )
+ return generated.questions
 
 
 def generate_quiz_for(concept_name: str) -> list[dict]: