ce1c0775ab
**Prepared sessions.** Most of this existed: unanswered first, weakest topic next, wrong-before-right after that, all scaled by what share of the real paper each topic carries. What it could not do was change with time, say anything about itself, or be reached without filling in a form. Evidence now decays on a thirty-day half-life. Exponential rather than a fixed window because memory has a slope, not a cliff — under a window, 29 days counts fully and 31 counts for nothing — and because it is memoryless, so an answer's weight does not shift when unrelated questions are answered, which is what lets the preview stay a valid forecast. Spring is worth an eighth of last week. Two things decay: a question's recall probability, drifting towards even rather than past it, so an old right answer becomes eligible rather than wrong; and a topic's accuracy, against a prior of two "no idea" answers, which fixes "right once, known forever". Strict unanswered-first meant that on a bank of 2,900 nothing was ever recycled — spaced repetition existed and was unreachable. Review now takes up to two fifths of a session. And the damping that spread the picks across topics was applied only to seen material, so a learner with no history was handed the heaviest domain entire instead of a spread; that was live. The plan is the product. It is computed, shown, and then the session is built from that plan's own ids and the plan returned with it, so the two cannot differ; every figure in it is a tally over the chosen questions rather than a forecast. No model touches the ranking — a learner asking "why these twenty" has to get the same answer twice. **Vision.** The proxy's own `/model/info` says which models can see, so nothing is hard-coded: 77 report yes, 11 no, and 328 say nothing at all, which means absent rather than incapable — so those are asked once with an 8px PNG and the refusal cached. The deployment's main model turns out not to see, and questions carry figures the learner is looking at, so the tutor was answering about an image it had never been shown. It routes to a configured tool model now, folds the description back in as text saying plainly where it came from, and caches on the bytes because the same figure is re-sent every turn. Also fixed on the way: `article` was missing from the admin's task list, so article drafting always ran on the fallback model whatever an administrator chose; and `.jpx` stem images were sent as JPEG because `mimetypes` guesses that from the name, so the provider rejected them two hops later. An administrator must pick a tool model in Settings → AI models. Until then the tutor says a figure exists that nothing could read, rather than describing one it cannot see. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01TqXevQJhxFrM7jJg82cgZN
367 lines
18 KiB
Python
367 lines
18 KiB
Python
"""The prepared session: the arithmetic, and whether the plan tells the truth.
|
||
|
||
The test that matters here is the last one. A plan that does not describe the
|
||
session it starts is worse than no plan, so the account handed to the learner
|
||
is checked against the questions actually put in the quiz — not against the
|
||
ranking that produced them, which would only prove the code agrees with itself.
|
||
"""
|
||
import os
|
||
|
||
os.environ.setdefault("DATABASE_URL", "sqlite:///:memory:")
|
||
|
||
import sys
|
||
import unittest
|
||
from datetime import datetime, timedelta
|
||
from decimal import Decimal
|
||
from types import ModuleType
|
||
from unittest.mock import patch
|
||
|
||
from fastapi import FastAPI
|
||
from fastapi.testclient import TestClient
|
||
from sqlalchemy import create_engine
|
||
from sqlalchemy.orm import Session
|
||
from sqlalchemy.pool import StaticPool
|
||
|
||
import test_quiz_builder # noqa: F401 — imports every model, so the metadata resolves
|
||
from app.database import Base, get_db
|
||
from app.models.attempt import AttemptAnswer, QuizAttempt
|
||
from app.models.exam import BlueprintCategoryLink, Exam, ExamBlueprint, QuestionExamLink
|
||
from app.models.question import Question
|
||
from app.models.question_category import QuestionCategory
|
||
from app.models.quiz import Quiz
|
||
from app.models.quiz_question_link import QuizQuestionLink
|
||
from app.models.user import User
|
||
from app.services import prepared_session
|
||
from app.services.quiz_builder import (DUE_RECALL, EVIDENCE_HALF_LIFE_DAYS, MAX_REVIEW_SHARE,
|
||
NEUTRAL_RECALL, CandidateRanking, recall_probability,
|
||
recency_weight)
|
||
from app.utils.auth import get_current_user
|
||
|
||
with patch.dict(sys.modules, {"app.services.quiz_service": ModuleType("app.services.quiz_service")}):
|
||
from app.routers import questions
|
||
|
||
#: Cardiology is filed a level down, which is where the plan's grouping has to
|
||
#: reach up from; the other two disciplines hold their questions directly.
|
||
CARDIOLOGY, CONGENITAL, RHEUMATOLOGY, DERMATOLOGY = 1, 2, 3, 4
|
||
|
||
|
||
class DecayTests(unittest.TestCase):
|
||
"""No database: the curve on its own."""
|
||
|
||
def test_evidence_halves_once_a_half_life(self):
|
||
self.assertAlmostEqual(recency_weight(0), 1.0)
|
||
self.assertAlmostEqual(recency_weight(EVIDENCE_HALF_LIFE_DAYS), 0.5)
|
||
self.assertAlmostEqual(recency_weight(3 * EVIDENCE_HALF_LIFE_DAYS), 0.125)
|
||
# An answer dated in the future is a clock disagreement, not evidence
|
||
# worth more than a fresh one.
|
||
self.assertAlmostEqual(recency_weight(-40), 1.0)
|
||
|
||
def test_both_outcomes_decay_towards_a_coin_flip_rather_than_past_it(self):
|
||
fresh_hit, old_hit = recall_probability(True, 0), recall_probability(True, 365)
|
||
fresh_miss, old_miss = recall_probability(False, 0), recall_probability(False, 365)
|
||
self.assertGreater(fresh_hit, old_hit)
|
||
self.assertLess(fresh_miss, old_miss)
|
||
# Time turns a right answer into "no idea", never into a wrong one.
|
||
self.assertAlmostEqual(old_hit, NEUTRAL_RECALL, places=2)
|
||
self.assertAlmostEqual(old_miss, NEUTRAL_RECALL, places=2)
|
||
self.assertGreater(old_hit, NEUTRAL_RECALL)
|
||
self.assertLess(old_miss, NEUTRAL_RECALL)
|
||
|
||
def test_a_recent_miss_outranks_an_older_one(self):
|
||
# The ranking is worth (1 − recall), so lower recall is picked first.
|
||
self.assertLess(recall_probability(False, 7), recall_probability(False, 90))
|
||
|
||
def test_a_correct_answer_comes_back_round_once_it_is_old_enough(self):
|
||
self.assertGreater(recall_probability(True, 3), DUE_RECALL)
|
||
self.assertLess(recall_probability(True, 90), DUE_RECALL)
|
||
# And a wrong answer is due from the moment it is given.
|
||
self.assertLess(recall_probability(False, 0), DUE_RECALL)
|
||
|
||
|
||
class Bank(unittest.TestCase):
|
||
"""Thirty questions over three disciplines, weighted 60 / 10 / 30."""
|
||
|
||
NOW = datetime(2026, 6, 1)
|
||
|
||
def setUp(self):
|
||
self.engine = create_engine("sqlite://", connect_args={"check_same_thread": False},
|
||
poolclass=StaticPool)
|
||
Base.metadata.create_all(self.engine)
|
||
self.db = Session(self.engine)
|
||
self.db.add(Exam(id=1, slug="boards", name="Pediatrics Boards", is_active=1))
|
||
self.user = User(id=1, name="Learner", email="l@example.test",
|
||
hashed_password="unused", active_exam_id=1)
|
||
self.db.add(self.user)
|
||
self.db.add_all([
|
||
QuestionCategory(id=CARDIOLOGY, name="Cardiology", user_id=1),
|
||
QuestionCategory(id=CONGENITAL, name="Congenital", parent_id=CARDIOLOGY, user_id=1),
|
||
QuestionCategory(id=RHEUMATOLOGY, name="Rheumatology", user_id=1),
|
||
QuestionCategory(id=DERMATOLOGY, name="Dermatology", user_id=1),
|
||
])
|
||
self.db.add(Quiz(id=1, title="Origin", user_id=1, is_published=1))
|
||
self.db.flush()
|
||
self.of_category = {}
|
||
for qid in range(1, 31):
|
||
category = CONGENITAL if qid <= 12 else RHEUMATOLOGY if qid <= 21 else DERMATOLOGY
|
||
self.of_category[qid] = category
|
||
self.db.add(Question(id=qid, question_category_id=category, user_id=1,
|
||
question_text=f"Q{qid}", question_type="mcq",
|
||
options=["a", "b"], correct_answer="a"))
|
||
self.db.add(QuestionExamLink(question_id=qid, exam_id=1))
|
||
self.db.add_all([
|
||
ExamBlueprint(id=1, exam_id=1, code="1", title="Cardiology", weight=Decimal("60")),
|
||
ExamBlueprint(id=2, exam_id=1, code="2", title="Rheumatology", weight=Decimal("10")),
|
||
ExamBlueprint(id=3, exam_id=1, code="3", title="Dermatology", weight=Decimal("30")),
|
||
])
|
||
self.db.flush()
|
||
self.db.add_all([BlueprintCategoryLink(blueprint_id=1, category_id=CARDIOLOGY),
|
||
BlueprintCategoryLink(blueprint_id=2, category_id=RHEUMATOLOGY),
|
||
BlueprintCategoryLink(blueprint_id=3, category_id=DERMATOLOGY)])
|
||
self.db.commit()
|
||
|
||
app = FastAPI()
|
||
app.include_router(questions.router, prefix="/questions")
|
||
app.dependency_overrides[get_db] = lambda: self.db
|
||
app.dependency_overrides[get_current_user] = lambda: self.user
|
||
self.client = TestClient(app)
|
||
|
||
def tearDown(self):
|
||
self.client.close()
|
||
self.db.close()
|
||
self.engine.dispose()
|
||
|
||
def sitting(self, marks, days_ago=1, when=None):
|
||
"""One finished attempt. `marks` is {question id: was it right}."""
|
||
finished = (when or self.NOW) - timedelta(days=days_ago)
|
||
attempt = QuizAttempt(user_id=1, quiz_id=1, completed_at=finished, expired=0,
|
||
total_questions=len(marks), score=sum(1 for ok in marks.values() if ok))
|
||
self.db.add(attempt)
|
||
self.db.flush()
|
||
for question_id, was_correct in marks.items():
|
||
self.db.add(AttemptAnswer(attempt_id=attempt.id, question_id=question_id,
|
||
is_correct=was_correct, user_answer="a" if was_correct else "b"))
|
||
self.db.commit()
|
||
return attempt
|
||
|
||
def ranking(self, now=None):
|
||
return CandidateRanking(self.db, self.user, now=now or self.NOW)
|
||
|
||
def plan(self, count=None, now=None):
|
||
return prepared_session.prepare_session(self.db, self.user, count, now or self.NOW)
|
||
|
||
|
||
class SelectionTests(Bank):
|
||
def test_a_miss_last_week_is_recycled_before_a_miss_in_the_spring(self):
|
||
# Same discipline, so the topic damping and the blueprint weight are
|
||
# identical and only the dates can separate them.
|
||
self.sitting({1: False}, days_ago=120)
|
||
self.sitting({2: False}, days_ago=7)
|
||
# Three questions leaves one review slot; the recent miss should take it.
|
||
picked = self.ranking().select(3)
|
||
self.assertIn(2, picked)
|
||
self.assertNotIn(1, picked)
|
||
|
||
def test_a_topic_answered_right_once_does_not_count_as_known_forever(self):
|
||
self.sitting({qid: True for qid in range(1, 13)}, days_ago=1)
|
||
fresh = self.ranking()
|
||
# Answered, recently, and correctly: nothing to review here yet.
|
||
self.assertEqual([row for row in fresh.due if row[0] <= 12], [])
|
||
stale = self.ranking(now=self.NOW + timedelta(days=180))
|
||
self.assertEqual(len(stale.due), 12)
|
||
# And the topic's accuracy has drifted back towards unknown rather than
|
||
# staying at the hundred percent one good day bought it.
|
||
self.assertGreater(fresh.accuracy(CONGENITAL), 0.8)
|
||
self.assertLess(stale.accuracy(CONGENITAL), 0.6)
|
||
|
||
def test_review_takes_its_share_and_no_more(self):
|
||
self.sitting({qid: False for qid in range(1, 13)}, days_ago=3)
|
||
picked = self.ranking().select(10)
|
||
seen = [qid for qid in picked if qid <= 12]
|
||
self.assertEqual(len(seen), round(MAX_REVIEW_SHARE * 10))
|
||
self.assertEqual(len(picked), 10)
|
||
|
||
def test_nothing_left_unseen_still_fills_the_session(self):
|
||
self.sitting({qid: qid % 2 == 0 for qid in range(1, 31)}, days_ago=3)
|
||
picked = self.ranking().select(12)
|
||
self.assertEqual(len(picked), 12)
|
||
self.assertEqual(len(set(picked)), 12)
|
||
|
||
def test_selection_is_reproducible(self):
|
||
self.sitting({1: False, 13: True, 25: False}, days_ago=10)
|
||
self.assertEqual(self.ranking().select(15), self.ranking().select(15))
|
||
|
||
|
||
class LengthTests(Bank):
|
||
def test_the_default_holds_until_enough_sessions_are_finished(self):
|
||
self.sitting({qid: True for qid in range(1, 6)}, days_ago=20)
|
||
self.sitting({qid: True for qid in range(6, 11)}, days_ago=15)
|
||
self.assertEqual(prepared_session.habitual_length(self.db, self.user), (None, 2))
|
||
self.assertEqual(self.plan()["count"], prepared_session.DEFAULT_SESSION_LENGTH)
|
||
|
||
def test_length_becomes_the_learner_s_own_median_finished_session(self):
|
||
# Eight, eight, thirty: the median is eight and the mean is fifteen,
|
||
# which is a length this learner has never once sat.
|
||
self.sitting({qid: True for qid in range(1, 9)}, days_ago=30)
|
||
self.sitting({qid: True for qid in range(9, 17)}, days_ago=20)
|
||
self.sitting({qid: True for qid in range(1, 31)}, days_ago=10)
|
||
self.assertEqual(prepared_session.habitual_length(self.db, self.user), (8, 3))
|
||
plan = self.plan()
|
||
self.assertEqual(plan["count"], 8)
|
||
self.assertIn("You usually finish 8 questions", plan["length_reason"])
|
||
|
||
def test_an_unfinished_or_expired_sitting_is_not_a_length(self):
|
||
attempt = self.sitting({qid: True for qid in range(1, 4)}, days_ago=5)
|
||
attempt.expired = 1
|
||
self.db.commit()
|
||
self.assertEqual(prepared_session.habitual_length(self.db, self.user), (None, 0))
|
||
|
||
def test_the_offered_length_never_exceeds_what_is_left(self):
|
||
for qid in range(6, 31):
|
||
self.db.delete(self.db.get(Question, qid))
|
||
self.db.commit()
|
||
plan = self.plan()
|
||
self.assertEqual(plan["count"], 5)
|
||
self.assertIn("everything left in your bank", plan["length_reason"])
|
||
|
||
def test_a_length_the_learner_asks_for_is_honored_and_said_to_be_theirs(self):
|
||
plan = self.plan(count=7)
|
||
self.assertEqual(plan["count"], 7)
|
||
self.assertEqual(plan["length_reason"], "7 questions, your choice.")
|
||
|
||
|
||
class PlanTests(Bank):
|
||
def rows(self, plan):
|
||
return {row["name"]: row for row in plan["topics"]}
|
||
|
||
def test_a_learner_with_no_history_is_told_so_and_gets_the_blueprint_spread(self):
|
||
plan = self.plan(count=10)
|
||
self.assertEqual(plan["basis"], "cold_start")
|
||
self.assertIn("haven't finished a session yet", plan["summary"])
|
||
self.assertIn("exam blueprint", plan["summary"])
|
||
self.assertEqual(plan["review_count"], 0)
|
||
rows = self.rows(plan)
|
||
# 60 / 10 / 30 of ten questions, which is what the board publishes.
|
||
self.assertEqual(rows["Cardiology"]["count"], 6)
|
||
self.assertEqual(rows["Rheumatology"]["count"], 1)
|
||
self.assertEqual(rows["Dermatology"]["count"], 3)
|
||
# And the reason says the blueprint, not a weakness nobody has measured.
|
||
self.assertEqual(rows["Cardiology"]["reason"], "Worth 60% of the exam")
|
||
for row in plan["topics"]:
|
||
self.assertIsNone(row["accuracy"])
|
||
|
||
def test_a_learner_with_one_topic_answered_is_told_which_and_how_badly(self):
|
||
self.sitting({qid: False for qid in range(1, 9)}, days_ago=4)
|
||
plan = self.plan(count=10)
|
||
self.assertEqual(plan["basis"], "personalized")
|
||
rows = self.rows(plan)
|
||
self.assertRegex(rows["Cardiology"]["reason"], r"^Weak area — \d+% correct so far$")
|
||
self.assertLess(rows["Cardiology"]["accuracy"], 30)
|
||
# Cardiology is both the weakest and the heaviest, so it leads — but the
|
||
# damping keeps the other two disciplines in the session.
|
||
self.assertEqual(plan["topics"][0]["name"], "Cardiology")
|
||
self.assertGreater(len(plan["topics"]), 1)
|
||
# Nothing has been measured about dermatology, and the plan says so
|
||
# rather than inventing a figure for it.
|
||
self.assertIsNone(rows["Dermatology"]["accuracy"])
|
||
self.assertEqual(rows["Dermatology"]["reason"], "Not attempted yet, worth 30% of the exam")
|
||
|
||
def test_a_topic_last_seen_months_ago_is_named_as_due_not_as_a_weakness(self):
|
||
# Everything right, but long enough ago that it is worth checking. The
|
||
# prior would drag the accuracy under the weak threshold; the evidence
|
||
# gate is what stops "you got all of these right" reading as "weak".
|
||
self.sitting({qid: True for qid in range(13, 22)}, days_ago=150)
|
||
self.sitting({qid: True for qid in range(22, 25)}, days_ago=2)
|
||
self.sitting({qid: True for qid in range(25, 28)}, days_ago=1)
|
||
rows = self.rows(self.plan(count=12))
|
||
self.assertIn("Rheumatology", rows)
|
||
self.assertRegex(rows["Rheumatology"]["reason"], r"^Due for review — last answered \d+ months ago$")
|
||
|
||
def test_the_counts_in_the_plan_add_up_to_the_session(self):
|
||
self.sitting({qid: qid % 3 == 0 for qid in range(1, 16)}, days_ago=45)
|
||
plan = self.plan(count=14)
|
||
self.assertEqual(sum(row["count"] for row in plan["topics"]), plan["count"])
|
||
self.assertEqual(plan["new_count"] + plan["review_count"], plan["count"])
|
||
for row in plan["topics"]:
|
||
self.assertEqual(row["new_count"] + row["review_count"], row["count"])
|
||
|
||
|
||
class CommittedPlanTests(Bank):
|
||
"""The plan has to describe the session it starts."""
|
||
|
||
def start(self, **body):
|
||
response = self.client.post("/questions/builder/prepared", json=body)
|
||
self.assertEqual(response.status_code, 200, response.text)
|
||
return response.json()
|
||
|
||
def questions_in(self, quiz_id):
|
||
return [row[0] for row in self.db.query(QuizQuestionLink.question_id).filter(
|
||
QuizQuestionLink.quiz_id == quiz_id).order_by(QuizQuestionLink.position).all()]
|
||
|
||
def test_the_plan_describes_the_questions_actually_put_in_the_session(self):
|
||
self.sitting({qid: qid % 4 == 0 for qid in range(1, 19)}, days_ago=6)
|
||
self.sitting({qid: True for qid in range(19, 25)}, days_ago=200)
|
||
self.sitting({25: False, 26: False}, days_ago=2)
|
||
created = self.start(count=16, title="Prepared")
|
||
plan = created["plan"]
|
||
asked = self.questions_in(created["id"])
|
||
|
||
self.assertEqual(len(asked), 16)
|
||
self.assertEqual(created["questions_count"], 16)
|
||
self.assertEqual(plan["count"], 16)
|
||
|
||
# Every claim the plan makes, recounted from the questions themselves.
|
||
by_discipline = {}
|
||
for question_id in asked:
|
||
leaf = self.of_category[question_id]
|
||
top = CARDIOLOGY if leaf == CONGENITAL else leaf
|
||
by_discipline[top] = by_discipline.get(top, 0) + 1
|
||
named = {CARDIOLOGY: "Cardiology", RHEUMATOLOGY: "Rheumatology", DERMATOLOGY: "Dermatology"}
|
||
self.assertEqual({row["name"]: row["count"] for row in plan["topics"]},
|
||
{named[key]: value for key, value in by_discipline.items()})
|
||
|
||
seen_before = {row[0] for row in self.db.query(AttemptAnswer.question_id).all()}
|
||
self.assertEqual(plan["review_count"], len([q for q in asked if q in seen_before]))
|
||
self.assertEqual(plan["new_count"], len([q for q in asked if q not in seen_before]))
|
||
for row in plan["topics"]:
|
||
top = next(key for key, name in named.items() if name == row["name"])
|
||
here = [q for q in asked
|
||
if (CARDIOLOGY if self.of_category[q] == CONGENITAL else self.of_category[q]) == top]
|
||
self.assertEqual(row["review_count"], len([q for q in here if q in seen_before]))
|
||
self.assertEqual(row["new_count"], len([q for q in here if q not in seen_before]))
|
||
# And every topic named gives its one reason.
|
||
for row in plan["topics"]:
|
||
self.assertTrue(row["reason"])
|
||
|
||
def test_the_preview_says_what_starting_it_will_give(self):
|
||
self.sitting({qid: qid % 2 == 0 for qid in range(1, 20)}, days_ago=9)
|
||
preview = self.client.get("/questions/builder/prepared", params={"count": 12})
|
||
self.assertEqual(preview.status_code, 200, preview.text)
|
||
forecast = preview.json()
|
||
# The preview is for reading; it must not hand out the stems' ids.
|
||
self.assertNotIn("question_ids", forecast)
|
||
started = self.start(count=12)["plan"]
|
||
self.assertEqual(forecast, started)
|
||
|
||
def test_the_preview_writes_nothing(self):
|
||
before = self.db.query(Quiz).count()
|
||
self.client.get("/questions/builder/prepared")
|
||
self.assertEqual(self.db.query(Quiz).count(), before)
|
||
|
||
def test_changing_the_length_replans_rather_than_padding_the_old_plan(self):
|
||
self.sitting({qid: False for qid in range(1, 10)}, days_ago=5)
|
||
short, long = self.start(count=6)["plan"], self.start(count=24)["plan"]
|
||
self.assertEqual(short["count"], 6)
|
||
self.assertEqual(long["count"], 24)
|
||
self.assertEqual(sum(row["count"] for row in long["topics"]), 24)
|
||
|
||
def test_without_a_study_objective_it_refuses_rather_than_guessing(self):
|
||
self.user.active_exam_id = None
|
||
self.db.commit()
|
||
for call in (self.client.get("/questions/builder/prepared"),
|
||
self.client.post("/questions/builder/prepared", json={})):
|
||
self.assertEqual(call.status_code, 400, call.text)
|
||
self.assertIn("studying for", call.json()["detail"])
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|