Spike readiness: AI gets its own slice of threads, cheaper hot paths, classroom-friendly login

- /api/ai/ask: at most AI_MAX_CONCURRENT_PER_WORKER (12 of 32 threads) answers per worker;
  beyond that 503 assistant_busy + Retry-After before any quota is reserved. Without it a
  burst of questions (3-40 s each) took every thread and the map and quizzes queued behind.
  The slot is held until the stream is actually finished or closed.
- Answer submission 21 -> 14 queries, 15.4 -> 11.0 ms: one query for all choices, locked
  get_or_create for node progress and streak, and today's streak recorded once per day
  via the cache instead of a transaction per answer.
- /api/home 12.4 -> 8.6 ms: the user's completions are read by user_id instead of an IN list
  of ~600 lesson ids; the dict is returned directly (serializer stays the Swagger contract,
  HomeMatchesItsSerializerTests keeps them equal).
- otp_login per-IP 30 -> 300/hour: a class of 30 on one school Wi-Fi filled it; guessing is
  bounded per code and per phone.
- /api/health aiConcurrency.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

ai_tutor/concurrency.py

@@ -0,0 +1,76 @@
+"""
+سقفِ جواب‌های هم‌زمانِ دستیار در هر worker (AI_MAX_CONCURRENT_PER_WORKER).
+
+جوابِ دستیار یک نخِ gunicorn را تا پایانِ کار (۳ تا ۴۰ ثانیه، منتظرِ Qdrant و مدل) نگه می‌دارد. بدونِ
+سقف، اگر چند ده دانش‌آموز با هم سوال بپرسند همه‌ی نخ‌های worker گرفته می‌شوند و نقشه، آزمون و ثبتِ
+جواب (که هرکدام چند میلی‌ثانیه‌اند) پشتِ آن‌ها صف می‌کشند: یک بخشِ کند کلِ اپ را می‌خواباند. این همان
+الگوی bulkhead است: هر بخش سهمِ جدای خودش را دارد. درخواستی که جا نگیرد فوراً ۵۰۳ می‌گیرد.
+
+سمافور مالِ همین پروسه است (هر worker جدا). هم‌زمانیِ واقعی زیرِ بارِ gunicorn با مدلِ واقعی آزموده نشده؛
+تست‌ها رفتارِ گرفتن/پس‌دادن را در مسیرِ JSON و استریم نگه می‌دارند (ai_tutor/tests_concurrency.py).
+"""
+import threading
+
+from django.conf import settings
+
+_lock = threading.Lock()
+_in_flight = 0
+
+
+def limit() -> int:
+    return settings.AI_MAX_CONCURRENT_PER_WORKER
+
+
+def try_acquire() -> bool:
+    """بی‌انتظار: یا جا هست و گرفته می‌شود، یا False. انتظار نخ را همان‌طور نگه می‌داشت که می‌خواهیم نشود."""
+    global _in_flight
+    with _lock:
+        if _in_flight >= limit():
+            return False
+        _in_flight += 1
+        return True
+
+
+def release() -> None:
+    global _in_flight
+    with _lock:
+        if _in_flight <= 0:
+            # پس‌دادنِ بی‌گرفتن یعنی باگ در جفت‌شدنِ گرفتن/پس‌دادن؛ شمارنده منفی نمی‌شود تا سقف بی‌اثر نشود.
+            raise RuntimeError('ai concurrency slot released more times than acquired')
+        _in_flight -= 1
+
+
+def in_flight() -> int:
+    return _in_flight
+
+
+class ReleaseWhenDone:
+    """
+    محتوای استریم را می‌پیچد تا جا وقتی پس داده شود که استریم واقعاً تمام شد: یا با پایانِ خواندن،
+    یا با close() که جنگو هنگامِ بستنِ پاسخ صدا می‌زند (کاربر وسطِ جواب صفحه را بست، یا پاسخ اصلاً
+    خوانده نشد). generatorی که شروع نشده با close() بلوکِ finallyاش را اجرا نمی‌کند؛ برای همین
+    پس‌دادن این‌جاست، نه داخلِ generator.
+    """
+
+    def __init__(self, iterable):
+        self._iterable = iterable
+        self._released = False
+
+    def _release_once(self):
+        if not self._released:
+            self._released = True
+            release()
+
+    def __iter__(self):
+        try:
+            yield from self._iterable
+        finally:
+            self._release_once()
+
+    def close(self):
+        try:
+            close = getattr(self._iterable, 'close', None)
+            if close is not None:
+                close()
+        finally:
+            self._release_once()

ai_tutor/tests_concurrency.py

@@ -0,0 +1,92 @@
+"""
+سقفِ جواب‌های هم‌زمانِ دستیار (ai_tutor/concurrency.py): رگبارِ سوال نباید همه‌ی نخ‌های worker را بگیرد.
+هم‌زمانیِ واقعیِ چند نخ با مدلِ واقعی آزموده نشده؛ این‌جا جا را با گرفتنِ دستی پر می‌کنیم.
+"""
+from unittest import mock
+
+from django.core.cache import cache
+from django.core.signals import request_finished
+from django.db import close_old_connections
+from django.test import TestCase, override_settings
+
+from accounts.authentication import issue_access_token
+from accounts.models import User
+from ai_tutor import concurrency
+from ai_tutor.models import Message
+from content.models import Grade
+
+
+@override_settings(AI_MAX_CONCURRENT_PER_WORKER=1)
+class AskConcurrencyCapTests(TestCase):
+    def setUp(self):
+        cache.clear()
+        grade = Grade.objects.create(title='پایه نهم', icon='', color='')
+        self.user = User.objects.create(phone='09120007801', grade_id=grade.id)
+        self.assertEqual(concurrency.in_flight(), 0)
+
+    def tearDown(self):
+        self.assertEqual(concurrency.in_flight(), 0, 'هر تست باید جا را پس بدهد')
+
+    def _ask(self, **extra):
+        return self.client.post(
+            '/api/ai/ask', {'userQuestion': 'توان چیست؟', **extra}, content_type='application/json',
+            HTTP_AUTHORIZATION=f'Bearer {issue_access_token(self.user.id)}',
+        )
+
+    def _rag(self):
+        return (
+            mock.patch('rag.services.config.rag_is_configured', return_value=True),
+            mock.patch('rag.services.retrieval.retrieve_context',
+                       return_value=([{'title': 'فصل', 'summary': 'متن'}], {'gradeTitle': 'نهم'})),
+            mock.patch('rag.services.generation.generate_answer', return_value=('جواب', [], {})),
+            mock.patch('rag.services.generation.stream_answer', side_effect=lambda *a, **k: iter(['جو', 'اب'])),
+        )
+
+    def test_full_means_503_with_retry_after_and_no_quota_used(self):
+        self.assertTrue(concurrency.try_acquire())  # یک سوالِ دیگر در حالِ جواب است
+        try:
+            with self.assertLogs('ai_tutor.views', level='WARNING'):
+                response = self._ask()
+        finally:
+            concurrency.release()
+        self.assertEqual(response.status_code, 503)
+        self.assertEqual(response.json()['detail'], 'دستیار الان خیلی شلوغه؛ چند ثانیه‌ی دیگه دوباره بپرس.')
+        self.assertEqual(response['Retry-After'], '10')
+        self.assertFalse(Message.objects.exists())  # نه پیامی ساخته شد نه سهمیه‌ای رزرو شد
+
+    def test_json_answer_gives_the_slot_back(self):
+        a, b, c, d = self._rag()
+        with a, b, c, d:
+            self.assertEqual(self._ask().status_code, 200)
+            self.assertEqual(concurrency.in_flight(), 0)
+            self.assertEqual(self._ask().status_code, 200)  # سقفِ ۱: دومی هم جا گرفت
+
+    def test_a_failed_answer_gives_the_slot_back(self):
+        a, b, _, d = self._rag()
+        failing = mock.patch('rag.services.generation.generate_answer', side_effect=RuntimeError('provider'))
+        with a, b, failing, d, self.assertLogs('ai_tutor.views', level='ERROR'):
+            self.assertEqual(self._ask().status_code, 503)
+
+    def test_stream_holds_the_slot_until_it_is_read(self):
+        a, b, c, d = self._rag()
+        with a, b, c, d:
+            response = self._ask(stream=True)
+            self.assertEqual(concurrency.in_flight(), 1)  # هنوز خوانده نشده
+            body = b''.join(response.streaming_content).decode()  # کلاینتِ تست پایانش close() می‌زند
+        self.assertIn('event: done', body)
+
+    def test_a_stream_that_is_never_read_still_gives_the_slot_back(self):
+        a, b, c, d = self._rag()
+        with a, b, c, d:
+            response = self._ask(stream=True)
+            # کاربر پیش از اولین تکه رفت. سیگنالِ request_finished در TestCase اتصالِ تستی را می‌بندد؛ کلاینتِ
+            # تست هم برای پاسخِ استریم همین‌طور جدایش می‌کند.
+            request_finished.disconnect(close_old_connections)
+            try:
+                response.close()
+            finally:
+                request_finished.connect(close_old_connections)
+
+    def test_release_without_acquire_is_a_loud_bug(self):
+        with self.assertRaises(RuntimeError):
+            concurrency.release()

ai_tutor/views.py

@@ -17,7 +17,7 @@ from rest_framework.views import APIView
 
 from accounts.models import User as AccountUser
 from config.db import release_db_connection
-from . import provider_status
+from . import concurrency, provider_status
 from .models import AskIdempotency, Conversation, Message, MessageFeedback
 from .serializers import (
     QuotaResponseSerializer,
@@ -256,6 +256,15 @@ def generate_tutor_answer(
     return answer, follow_ups
 
 
+class AIBusy(APIException):
+    """همه‌ی جاهای هم‌زمانِ دستیار در این worker پرند (ai_tutor/concurrency.py). سهمیه دست نخورده."""
+    status_code = 503
+    default_detail = 'دستیار الان خیلی شلوغه؛ چند ثانیه‌ی دیگه دوباره بپرس.'
+    default_code = 'assistant_busy'
+    # هندلرِ DRF از `wait` هدرِ Retry-After می‌سازد.
+    wait = 10
+
+
 class AskAIView(APIView):
     """POST /api/ai/ask — سوال از AI (استریم اختیاری؛ در این فاز استریم پیاده نشده)"""
 
@@ -282,6 +291,24 @@ class AskAIView(APIView):
         tags=['AI Tutor (ai_tutor)'],
     )
     def post(self, request):
+        # پیش از هر کارِ دیگری، حتی رزروِ سهمیه: درخواستِ ردشده نباید از سهمیه‌ی روزانه کم کند.
+        if not concurrency.try_acquire():
+            logger.warning('ai_busy user_id=%s in_flight=%s limit=%s',
+                           request.user.id, concurrency.in_flight(), concurrency.limit())
+            raise AIBusy()
+        try:
+            response = self._ask(request)
+        except BaseException:
+            concurrency.release()
+            raise
+        if isinstance(response, StreamingHttpResponse):
+            # جا تا پایانِ استریم دستِ همین درخواست می‌ماند (ReleaseWhenDone).
+            response.streaming_content = concurrency.ReleaseWhenDone(response.streaming_content)
+        else:
+            concurrency.release()
+        return response
+
+    def _ask(self, request):
         serializer = AskRequestSerializer(data=request.data)
         serializer.is_valid(raise_exception=True)
         v = serializer.validated_data