tutor answers step by step in plain text, islands finish, and the audit's money and login holes are closed

Tutor (prompt tutor-v3-stepwise-plain): plain text instead of LaTeX/markdown,
with rag/services/mathtext.py converting whatever still slips through; full
step-by-step solutions with a check before the final answer; three tones by
grade; the last 6 messages of the conversation go to the model and follow-ups
are no longer served from another user's cache; book solutions the checker
disputes are kept out of the context; a questionId gives the tutor the
practice question and the student's wrong answer. /api/health reports when
the AI credit runs out. rag_model_benchmark compares chat models on verified
book problems (run on the server; credit ran out mid-run).

Map: only lessons that have questions count toward finishing an island,
lessonIds[0] / nextLessonId is always the next lesson to play, and a finished
island stays finished. The Seyar characters (sigma/delta/pi) are seeded.

Audit fixes, each reproduced before fixing:
- the plan is stored on the order, so paying a monthly order after starting
  a yearly checkout no longer grants a year
- opening checkout no longer demotes an active subscription; renewals extend
  from the current expiry; expired subscriptions are not reported active
- one discount code per order and only before the gateway; a 100% code
  completes the purchase; lost Zarinpal confirmations are re-verified
- the unused /api/payment/webhook "mark paid" endpoint is removed
- XP counts only answers recorded on the server
- 6-digit OTP, each code dies after 5 wrong guesses; POST /api/auth/logout
- a guest who signs up keeps their account and progress (test phones excluded)
- affiliate commission on the first real payment; guests excluded
- parents with several children no longer get a 500
- SQLite runs transactions IMMEDIATE, since it ignores select_for_update
- admin login is rate-limited; display names drop phone numbers and links;
  userQuestion is capped at 2000 characters
- server-side heart enforcement exists behind HEARTS_ENFORCED (off until the
  frontend handles 403 no_hearts)

Not done on purpose: rotating the server's secret key and moving to Postgres.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
2026-09-25 06:24:13 +03:30
parent fbdb9495ee
commit 42783f9ba4
56 changed files with 2792 additions and 326 deletions

302
rag/tests_tutor_quality.py Normal file
View File

@@ -0,0 +1,302 @@
"""
کیفیتِ پاسخِ دستیار (۲۰۲۶-۰۹-۲۵): متنِ ساده به‌جای LaTeX، لحنِ سه‌سطحی، حلِ
مرحله‌به‌مرحله با وارسی، حافظه‌ی گفتگو، و کنار گذاشتنِ راه‌حل‌های غلطِ کتاب.
هر تست یکی از ایرادهایی را می‌پوشاند که روی جواب‌های واقعی دیده شد؛
توضیحِ هرکدام کنارِ خودِ کد است.
"""
from types import SimpleNamespace
from unittest.mock import patch
from django.core.cache import cache
from django.test import SimpleTestCase, TestCase, override_settings
from rest_framework.test import APIClient
from accounts.authentication import issue_access_token
from accounts.models import User
from .services.clients import reset_clients_cache
from .services.mathtext import StreamSanitizer, to_plain
RAG_ON = dict(
OPENAI_API_KEY='test-openai', QDRANT_API_KEY='test-qdrant', RAG_FORCE_MOCK=False,
)
class PlainMathTextTests(SimpleTestCase):
def test_inline_and_display_latex_become_unicode(self):
text = r'مشتق \(f(x)=x^{3}-6x^2+9x\) برابر است با:' '\n' r'\[ f\'(x) = 3x^2 - 12x + 9 \]'
self.assertEqual(to_plain(text), "مشتق f(x)=x³-6x²+9x برابر است با:\nf'(x) = 3x² - 12x + 9")
def test_fractions_keep_their_meaning(self):
# «1/2h» یعنی h/2 — مخرجِ چندنویسه‌ای باید پرانتز بگیرد.
self.assertEqual(to_plain(r'\(\frac{a}{2h}\)'), 'a/(2h)')
self.assertEqual(to_plain(r'\(\frac{x+1}{x-2}\)'), '(x+1)/(x-2)')
self.assertEqual(to_plain(r'\(\frac{\sqrt{3}}{2}\)'), '√3/2')
def test_primary_fractions_are_written_in_words(self):
# در کتابِ دبستان «۲/۵» یعنی دو و نیم؛ کسر نباید با «/» نوشته شود.
self.assertEqual(to_plain(r'\(\frac{3}{4}\)', fraction_word=True), '3 بر 4')
def test_roots_limits_degrees_and_symbols(self):
self.assertEqual(to_plain(r'\(\sqrt[3]{8} = 2\)'), '³√8 = 2')
self.assertEqual(to_plain(r'\(\lim_{x \to 0} \frac{\sin x}{x} = 1\)'), 'lim(x → 0) (sin x)/x = 1')
self.assertEqual(to_plain(r'\(30^\circ\)'), '30°')
self.assertEqual(to_plain(r'\(a \leq b \neq c \times d\)'), 'a ≤ b ≠ c × d')
self.assertEqual(to_plain('x^10 + y^2'), 'x¹⁰ + y²')
def test_set_braces_survive(self):
# مجموعه‌ها در ریاضیِ نهم با آکولاد نوشته می‌شوند.
self.assertEqual(to_plain('A = {1, 2, 3}'), 'A = {1, 2, 3}')
self.assertEqual(to_plain(r'\(A = \{1, 2\} \cup B\)'), 'A = {1, 2} ∪ B')
def test_markdown_is_removed(self):
text = '### مرحله ۱\n**نکته:** این مهم است\n| x | y |\n|---|---|\n| 1 | 2 |\n- اول'
self.assertEqual(to_plain(text), 'مرحله ۱\nنکته: این مهم است\nx | y\n1 | 2\n• اول')
def test_plain_text_passes_through(self):
text = '۱) دو طرف را بر ۲ تقسیم می‌کنیم.\n۲) x = 3\n✅ جواب: 3'
self.assertEqual(to_plain(text), text)
def test_stream_output_equals_batch_output_however_it_is_split(self):
text = (
'### حل\nاول \\(x^{2}\\) را حساب می‌کنیم:\n\\[\n\\frac{a+b}{2}\n\\]\n\n\n'
'| a | b |\n|---|---|\n| 1 | 2 |\n**تمام** \\(\\sqrt{16}=4\\)'
)
for step in (1, 2, 3, 7, 50):
sanitizer = StreamSanitizer()
out = ''.join(sanitizer.feed(text[i:i + step]) for i in range(0, len(text), step))
out += sanitizer.flush()
self.assertEqual(out, to_plain(text), f'step={step}')
class TutorPromptTests(SimpleTestCase):
def test_three_tiers_have_different_voices(self):
from rag.services.generation import system_prompt
primary, middle, upper = system_prompt('سوم'), system_prompt('هشتم'), system_prompt('یازدهم')
self.assertIn('دبستانی', primary)
self.assertIn('سیب', primary)
self.assertNotIn('دبستان', middle)
self.assertIn('دبیرستان', upper)
self.assertIn('دامنه', upper)
def test_every_prompt_demands_full_steps_a_check_and_plain_text(self):
from rag.services.generation import system_prompt
for grade in ('سوم', 'نهم', 'دوازدهم', ''):
prompt = system_prompt(grade)
self.assertIn('مرحله‌به‌مرحله', prompt)
self.assertIn('وارسی', prompt)
self.assertIn('✅ جواب', prompt)
self.assertIn('LaTeX', prompt)
def test_decimal_notation_follows_the_book_for_primary(self):
from rag.services.generation import system_prompt
self.assertIn('۲/۵ یعنی', system_prompt('ششم'))
self.assertIn('(3.5)', system_prompt('دهم'))
def test_history_goes_before_the_new_question(self):
from rag.services.generation import HISTORY_TURNS, build_messages
history = [{'role': 'user', 'text': f'سوال {i}'} if i % 2 == 0
else {'role': 'assistant', 'text': f'جواب {i}'} for i in range(10)]
messages = build_messages('نفهمیدم', [{'summary': 'متنِ کتاب'}], 'نهم', history)
self.assertEqual(messages[0]['role'], 'system')
self.assertEqual(len(messages), 1 + HISTORY_TURNS + 1)
self.assertEqual(messages[-2], {'role': 'assistant', 'content': 'جواب 9'})
self.assertTrue(messages[-1]['content'].endswith('سوالِ دانش‌آموز: نفهمیدم'))
self.assertIn('متنِ کتاب', messages[-1]['content'])
def test_reasoning_models_get_no_temperature(self):
from rag.services.generation import completion_kwargs
self.assertEqual(completion_kwargs('gpt-4o-mini'), {'temperature': 0.2})
with override_settings(OPENAI_CHAT_REASONING_EFFORT='low'):
self.assertEqual(completion_kwargs('o4-mini'), {'reasoning_effort': 'low'})
self.assertEqual(completion_kwargs('gpt-5-mini'), {'reasoning_effort': 'low'})
@override_settings(OPENAI_CHAT_MODEL='cheap', OPENAI_CHAT_MODEL_UPPER='strong')
def test_upper_grades_can_have_their_own_model(self):
from rag.services.generation import chat_model_for
self.assertEqual(chat_model_for('دوازدهم'), 'strong')
self.assertEqual(chat_model_for('نهم'), 'cheap')
self.assertEqual(chat_model_for(''), 'cheap')
@override_settings(**RAG_ON)
class GenerateAnswerSanitizesTests(TestCase):
def setUp(self):
reset_clients_cache()
def tearDown(self):
reset_clients_cache()
def _client(self, reply):
completions = SimpleNamespace(create=lambda **kw: SimpleNamespace(
choices=[SimpleNamespace(message=SimpleNamespace(content=reply))], usage=None))
return SimpleNamespace(chat=SimpleNamespace(completions=completions))
def test_latex_from_the_model_never_reaches_the_student(self):
from rag.services.generation import generate_answer
with patch('rag.services.clients.get_openai_client',
return_value=self._client(r'۱) \(x^2 = 4\)' '\n' r'✅ جواب: \(x = \pm 2\)')):
answer, _f, meta = generate_answer('x² = 4', None, 'نهم')
self.assertEqual(answer, '۱) x² = 4\n✅ جواب: x = ±2')
self.assertTrue(meta['sanitized'])
self.assertEqual(meta['promptVersion'], 'tutor-v3-stepwise-plain')
class DisputedSolutionTests(SimpleTestCase):
def _payload(self, verdict):
return {'generalSolution': 'روش', 'problem': 'مسئله', 'solution': 'راه‌حلِ کتاب',
'solution_verdict': verdict}
def test_a_solution_the_checker_disputes_is_not_shown_to_the_tutor(self):
from rag.services.retrieval import _payload_text
text = _payload_text(self._payload('disagrees'))
self.assertIn('مسئله', text)
self.assertNotIn('راه‌حلِ کتاب:', text)
def test_agreed_and_unchecked_solutions_still_go(self):
from rag.services.retrieval import _payload_text
for verdict in ('agrees', 'unjudgeable', None):
self.assertIn('راه‌حلِ کتاب: راه‌حلِ کتاب', _payload_text(self._payload(verdict)))
@override_settings(**RAG_ON, RAG_ANSWER_CACHE_TTL=300)
class ConversationMemoryTests(TestCase):
def setUp(self):
cache.clear()
reset_clients_cache()
self.client = APIClient()
def tearDown(self):
cache.clear()
def _user(self, phone):
user = User.objects.create(phone=phone, is_guest=False)
client = APIClient()
client.credentials(HTTP_AUTHORIZATION=f'Bearer {issue_access_token(user.id)}')
return client
def _ask(self, client, text, conversation_id=None):
body = {'userQuestion': text, 'subject': 'ریاضی', 'grade': 'نهم', 'lessonNumber': 4}
if conversation_id:
body['conversationId'] = conversation_id
return client.post('/api/ai/ask', body, format='json')
def test_follow_up_sees_the_conversation_and_searches_with_the_topic(self):
client = self._user('09120000901')
retrieve = patch('rag.services.retrieval.retrieve_context',
return_value=([], {'gradeTitle': 'نهم'}))
generate = patch('rag.services.generation.generate_answer',
return_value=('جوابِ توان', [], {}))
with retrieve as retrieve_mock, generate as generate_mock:
first = self._ask(client, '۲ به توان منفی ۳ چند است؟')
self._ask(client, 'نفهمیدم', first.json()['conversationId'])
history = generate_mock.call_args.kwargs['history']
self.assertEqual([t['text'] for t in history], ['۲ به توان منفی ۳ چند است؟', 'جوابِ توان'])
self.assertIn('توان منفی', retrieve_mock.call_args.args[0])
def test_a_follow_up_is_never_served_from_someone_elses_cache(self):
replies = iter(['جوابِ اولِ الف', 'جوابِ «نفهمیدم»ِ الف', 'جوابِ اولِ ب', 'جوابِ «نفهمیدم»ِ ب'])
retrieve = patch('rag.services.retrieval.retrieve_context',
return_value=([], {'gradeTitle': 'نهم'}))
generate = patch('rag.services.generation.generate_answer',
side_effect=lambda *a, **k: (next(replies), [], {}))
with retrieve, generate:
a, b = self._user('09120000902'), self._user('09120000903')
conv_a = self._ask(a, 'اتحاد چیست؟').json()['conversationId']
self._ask(a, 'نفهمیدم', conv_a)
conv_b = self._ask(b, 'تجزیه چیست؟').json()['conversationId']
second_b = self._ask(b, 'نفهمیدم', conv_b)
self.assertEqual(second_b.json()['answer'], 'جوابِ «نفهمیدم»ِ ب')
@override_settings(**RAG_ON)
class ProviderStateTests(TestCase):
"""اعتبارِ تمام‌شده باید در /api/health دیده شود، نه فقط در لاگ."""
def setUp(self):
cache.clear()
reset_clients_cache()
user = User.objects.create(phone='09120000904', is_guest=False)
self.client = APIClient()
self.client.credentials(HTTP_AUTHORIZATION=f'Bearer {issue_access_token(user.id)}')
def tearDown(self):
cache.clear()
def _ask(self, generate):
with patch('rag.services.retrieval.retrieve_context', return_value=([], {})), \
patch('rag.services.generation.generate_answer', **generate):
return self.client.post('/api/ai/ask', {'userQuestion': 'سلام'}, format='json')
def test_exhausted_credit_shows_up_in_health_and_clears_on_recovery(self):
quota = RuntimeError("Error code: 403 - insufficient user quota, remaining quota: $-0.001")
self.assertEqual(self._ask({'side_effect': quota}).status_code, 503)
health = self.client.get('/api/health').json()
self.assertEqual(health['integrations']['aiProvider']['state'], 'quota_exhausted')
self.assertTrue(any('GapGPT' in w for w in health['warnings']))
self.assertEqual(self._ask({'return_value': ('جواب', [], {})}).status_code, 200)
health = self.client.get('/api/health').json()
self.assertEqual(health['integrations']['aiProvider']['state'], 'ok')
def test_an_ordinary_network_error_does_not_claim_the_credit_is_gone(self):
self.assertEqual(self._ask({'side_effect': ConnectionError('timeout')}).status_code, 503)
health = self.client.get('/api/health').json()
self.assertEqual(health['integrations']['aiProvider']['state'], 'ok')
@override_settings(**RAG_ON)
class QuestionFocusTests(TestCase):
"""«چرا جوابم غلط بود؟» — دستیار باید همان سوال و جوابِ دانش‌آموز را ببیند."""
def setUp(self):
from content.models import Choice, Grade, Lesson, Node, Question, Unit
cache.clear()
grade = Grade.objects.create(title='پایه نهم', icon='', color='')
node = Node.objects.create(unit=Unit.objects.create(grade=grade, title='ف', order=1),
title='ن', order=1)
self.question = Question.objects.create(lesson=Lesson.objects.create(node=node, title='د'),
type='multiple_choice', prompt='۲ به توان منفی ۳؟')
self.right = Choice.objects.create(question=self.question, value='1/8', is_correct=True)
self.wrong = Choice.objects.create(question=self.question, value='-8', is_correct=False)
user = User.objects.create(phone='09120000905', is_guest=False, grade_id=grade.id)
self.client = APIClient()
self.client.credentials(HTTP_AUTHORIZATION=f'Bearer {issue_access_token(user.id)}')
def tearDown(self):
cache.clear()
def test_the_tutor_sees_the_question_and_the_wrong_answer(self):
self.client.post(f'/api/questions/{self.question.id}/answer',
{'selectedOptionId': self.wrong.id}, format='json')
with patch('rag.services.retrieval.retrieve_context', return_value=([], {})), \
patch('rag.services.generation.generate_answer',
return_value=('جواب', [], {})) as generate:
response = self.client.post('/api/ai/ask', {
'userQuestion': 'چرا غلط شد؟', 'questionId': str(self.question.id),
}, format='json')
self.assertEqual(response.status_code, 200)
focus = generate.call_args.kwargs['focus']
self.assertIn('۲ به توان منفی ۳؟', focus)
self.assertIn('-8 (غلط)', focus)
self.assertIn('جوابِ درست: 1/8', focus)
def test_an_overlong_question_is_refused_before_any_paid_call(self):
with patch('rag.services.generation.generate_answer') as generate:
response = self.client.post('/api/ai/ask', {'userQuestion': 'x' * 2001}, format='json')
self.assertEqual(response.status_code, 400)
generate.assert_not_called()