let a book be re-read without duplicating everything in it

ingestِ دوباره جایگزین نمی‌کرد: qdrant_point_id یک uuid4 تازه به ازای هر ردیف
است، پس خواندنِ دوباره‌ی یک صفحه نسخه‌ی تکراری کنارِ نسخه‌ی قدیمی می‌گذاشت و
جستجو نمی‌توانست تشخیص بدهد کدام کدام است.

--refresh {figures,unjudgeable,both,all} و --refresh-pages اضافه شدند.

سه تصمیمِ طراحی که عمدی‌اند:

۱. پاک‌سازی صفحه‌به‌صفحه است و درست پیش از خواندنِ همان صفحه، نه یک‌جا از اول.
   وسطِ تست یک crash همین را ثابت کرد: فقط یک صفحه خالی ماند (۲۷۹→۲۷۶) و بقیه‌ی
   کتاب دست‌نخورده. پاک‌سازیِ یک‌جا همان crash را به ۸۱ صفحه‌ی خالی تبدیل می‌کرد.

۲. اول Qdrant، بعد دیتابیس. شناسه‌ی نقطه فقط در ردیفِ DB است، پس ترتیبِ برعکس
   یعنی شکستِ Qdrant نقاط را برای همیشه بی‌صاحب می‌گذارد — نه قابلِ یافتن، نه
   قابلِ پاک کردن.

۳. order از Max گرفته می‌شود نه count: بازخوانی ردیف‌ها را از وسط پاک می‌کند و
   count عددی می‌داد که با orderهای موجود تصادم داشت.

تخمینِ هزینه حالا از اندازه‌گیریِ واقعیِ پنلِ صورتحساب می‌آید نه از جدولِ قیمت،
و وعده‌ی «با Batch API نصف» حذف شد چون GapGPT اصلاً /v1/batches ندارد.

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

data/textbooks/README.md

@@ -16,11 +16,15 @@
 
 | پایه | وضعیت |
 |---|---|
-| ششم، هفتم | از قبل روی Qdrant موجود است (ساخته‌ی تیم دیگر) |
-| نهم، دهم | PDF موجود است؛ ingest با `rag_ingest_book` |
-| هشتم | PDF موجود است، هنوز در `textbooks.py` ثبت نشده |
+| سوم، ششم، هفتم | روی Qdrant هست (ساخته‌ی تیم دیگر) — PDF نداریم جز هفتم |
+| هشتم، نهم، دهم | ingest شده؛ بازخوانیِ صفحه‌های شکل‌دار مانده |
+| یازدهم، دوازدهم | در نوبتِ ingest (۱۴۰۵/۰۶/۲۸) |
 | اول، دوم، چهارم، پنجم | PDF موجود است، خارج از اولویتِ فعلی |
 
+⚠️ کتابِ ریاضیِ یازدهم و دوازدهم چند رشته دارد (تجربی، ریاضی‌فیزیک، انسانی).
+نامِ فایل و `Textbook.grade` باید رشته را مشخص کند، وگرنه دو کتابِ متفاوت
+سرِ یک collection دعوا می‌کنند.
+
 ## نام‌گذاری
 
 نامِ فایل باید دقیقاً با `filename` در `rag/services/textbooks.py` یکی باشد،
@@ -42,6 +46,35 @@ python manage.py rag_ingest_book --grade نهم
 هر اجرا صفحه‌هایی را که قبلاً chunk ساخته‌اند رد می‌کند، پس قطع کردنش با
 Ctrl-C بی‌خطر است و اجرای دوباره هزینه‌ی رفته را دوباره نمی‌پردازد.
 
+## بازخوانیِ صفحه‌های معیوب
+
+کتاب‌هایی که پیش از افزوده شدنِ قاعده‌ی «self-contained» به پرامپت ingest
+شده‌اند، مسئله‌هایی دارند که به «شکلِ زیر» یا «جدولِ روبه‌رو» ارجاع می‌دهند
+بی‌آنکه محتوای آن شکل در متن آمده باشد. مدل آن شکل را **دیده بود**؛ فقط از او
+خواسته نشده بود بنویسدش. پس با خواندنِ دوباره‌ی همان صفحه‌ها جبران می‌شود.
+
+```bash
+# اول ببینید چند صفحه و چه هزینه‌ای — رایگان، هیچ فراخوانیِ مدلی ندارد
+python manage.py rag_ingest_book --grade نهم --refresh both --plan
+
+# فقط صفحه‌های ارجاع‌دهنده به شکل/جدول
+python manage.py rag_ingest_book --grade نهم --refresh figures
+
+# صفحه‌های مشخص (وقتی خودتان فهرست دارید)
+python manage.py rag_ingest_book --grade نهم --refresh-pages 18,44,60-65
+```
+
+سه نکته که باید بدانید:
+
+- **بازخوانی پاک می‌کند.** واحدِ کار «صفحه» است نه chunk، چون vision یک صفحه را
+  یک‌جا می‌خواند. پس chunkهای سالمِ همان صفحه هم دور ریخته و از نو ساخته
+  می‌شوند و `solution_verdict`شان پاک می‌شود. فرمان پیش از خرج کردن نشان
+  می‌دهد چند تای آن‌ها رأیِ «agrees» داشته‌اند و تأیید می‌گیرد (`--yes` رد می‌کند).
+- **بعدش `rag_verify_solutions` را دوباره اجرا کنید،** وگرنه chunkهای تازه
+  بی‌رأی می‌مانند و تولیدِ سؤال — که فقط `agrees` می‌خواند — نادیده‌شان می‌گیرد.
+- **پاک‌سازی صفحه‌به‌صفحه است،** درست پیش از خواندنِ همان صفحه. پس قطعِ وسطِ کار
+  (Ctrl-C، تمام شدنِ اعتبار) فقط یک صفحه را خالی می‌گذارد، نه کلِ کتاب را.
+
 ## افزودنِ کتابِ جدید
 
 در `rag/services/textbooks.py` یک `Textbook` بسازید با:

rag/management/commands/rag_ingest_book.py

@@ -13,7 +13,8 @@ from django.core.management.base import BaseCommand, CommandError
 
 from rag.services.clients import RagNotConfigured
 from rag.services.config import build_collection_name, rag_is_configured
-from rag.services.ingest import _done_pages, estimate_page_cost, ingest_book
+from rag.services.ingest import (_done_pages, estimate_page_cost, ingest_book,
+                                 pages_needing_refresh)
 from rag.services.textbooks import TEXTBOOKS, get_textbook
 from rag.models import IngestedPage, LessonChunk
 
@@ -33,6 +34,18 @@ class Command(BaseCommand):
                             help='صفحه‌های قبلاً ثبت‌شده را هم دوباره پردازش کن (نقطه‌ی تکراری می‌سازد).')
         parser.add_argument('--plan', action='store_true',
                             help='فقط نشان بده چه چیزی پردازش می‌شود؛ هیچ فراخوانیِ پولی نکن.')
+        parser.add_argument('--refresh', choices=['figures', 'unjudgeable', 'both', 'all'],
+                            default=None,
+                            help='صفحه‌های معیوب را دوباره بخوان: دادهٔ قبلیِ هر صفحه '
+                                 'درست پیش از خواندنِ دوباره پاک می‌شود. '
+                                 'figures=ارجاع به شکل/جدولِ نوشته‌نشده، '
+                                 'unjudgeable=رأیِ «قابلِ قضاوت نیست»، both=هر دو، '
+                                 'all=همه‌ی صفحه‌های فصل‌های انتخاب‌شده.')
+        parser.add_argument('--refresh-pages', default='',
+                            help='به‌جای انتخابِ خودکار، همین صفحه‌های PDF را بازخوانی کن '
+                                 '(مثلاً: 12,15,20-24).')
+        parser.add_argument('--yes', action='store_true',
+                            help='بدونِ پرسشِ تأیید ادامه بده (بازخوانی دادهٔ موجود را پاک می‌کند).')
 
     def handle(self, *args, **opts):
         subject, grade = opts['subject'], opts['grade']
@@ -54,8 +67,14 @@ class Command(BaseCommand):
         if not wanted:
             raise CommandError('هیچ فصلی با شماره‌های داده‌شده پیدا نشد.')
 
+        refresh = self._resolve_refresh(opts, collection, book, wanted)
+
         if opts['plan']:
-            return self._plan(book, collection, wanted)
+            return self._plan(book, collection, wanted, refresh)
+
+        if refresh and not opts['yes'] and not self._confirm(collection, refresh):
+            self.stdout.write('لغو شد؛ هیچ‌چیزی پاک یا خرج نشد.')
+            return
 
         if not rag_is_configured():
             raise CommandError(
@@ -70,7 +89,8 @@ class Command(BaseCommand):
                 subject=book.subject, grade=book.grade,
                 chapters=[c.number for c in wanted],
                 dpi=opts['dpi'], max_pages=opts['max_pages'],
-                resume=not opts['no_resume'], progress=self._progress,
+                resume=not opts['no_resume'], refresh_pages=refresh,
+                progress=self._progress,
             )
         except (RagNotConfigured, FileNotFoundError, ValueError) as exc:
             raise CommandError(str(exc)) from exc
@@ -82,6 +102,12 @@ class Command(BaseCommand):
             f"{result['pages_skipped']} صفحه از قبل موجود بود، "
             f"{result['chunks_created']} chunk ساخته شد."
         ))
+        if result.get('chunks_replaced'):
+            self.stdout.write(self.style.WARNING(
+                f"بازخوانی: {result['chunks_replaced']} chunkِ قدیمی پاک و با نسخه‌ی تازه "
+                f"جایگزین شد. رأیِ بازبینیِ این‌ها پاک شده — "
+                f"«python manage.py rag_verify_solutions» را دوباره اجرا کنید."
+            ))
         self.stdout.write(
             f"مصرف: {result['prompt_tokens']:,} توکن ورودی + "
             f"{result['completion_tokens']:,} خروجی  →  ≈ ${result['cost_usd']:.2f} "
@@ -94,34 +120,105 @@ class Command(BaseCommand):
             ))
 
     # ------------------------------------------------------------------
-    def _plan(self, book, collection, wanted):
+    @staticmethod
+    def _parse_pages(spec: str) -> set[int]:
+        """«12,15,20-24» → {12, 15, 20, 21, 22, 23, 24}"""
+        pages: set[int] = set()
+        for part in (p.strip() for p in spec.split(',') if p.strip()):
+            if '-' in part:
+                first, _, last = part.partition('-')
+                try:
+                    pages.update(range(int(first), int(last) + 1))
+                except ValueError as exc:
+                    raise CommandError(f'بازه‌ی نامعتبر در --refresh-pages: {part!r}') from exc
+            else:
+                try:
+                    pages.add(int(part))
+                except ValueError as exc:
+                    raise CommandError(f'شماره‌ی نامعتبر در --refresh-pages: {part!r}') from exc
+        return pages
+
+    def _resolve_refresh(self, opts, collection, book, wanted) -> set[int]:
+        """کدام صفحه‌ها بازخوانی شوند — محدود به فصل‌های انتخاب‌شده."""
+        if opts['refresh_pages']:
+            pages = self._parse_pages(opts['refresh_pages'])
+        elif opts['refresh'] == 'all':
+            # همه‌ی صفحه‌های فصل‌های این اجرا. انتخابِ هدفمند به سیگنال‌هایی
+            # تکیه می‌کند که اندازه‌گیری نشان داد نویزی‌اند (regexِ «شکل» هم
+            # «به شکل مخروطی» را می‌گیرد، و ۷۷٪ از رأی‌های unjudgeable از همان
+            # پیش‌فیلترِ گشاد آمده)، و اختلافِ هزینه‌اش ناچیز است.
+            pages = set()
+            for chapter in wanted:
+                first, last = book.pdf_pages(chapter)
+                pages.update(range(first, last + 1))
+        elif opts['refresh']:
+            mode = opts['refresh']
+            pages = pages_needing_refresh(
+                collection,
+                figures=mode in ('figures', 'both'),
+                unjudgeable=mode in ('unjudgeable', 'both'),
+            )
+        else:
+            return set()
+
+        # فقط صفحه‌هایی که داخلِ فصل‌های همین اجرا هستند؛ وگرنه --chapters
+        # بی‌اثر می‌شد و یک اجرای محدود کلِ کتاب را پاک می‌کرد.
+        in_scope: set[int] = set()
+        for chapter in wanted:
+            first, last = book.pdf_pages(chapter)
+            in_scope.update(p for p in pages if first <= p <= last)
+        return in_scope
+
+    def _confirm(self, collection, refresh) -> bool:
+        chunks = LessonChunk.objects.filter(
+            collection=collection, source_page__in=refresh)
+        verified = chunks.filter(solution_verdict='agrees').count()
+        self.stdout.write(self.style.WARNING(
+            f'\n⚠️  بازخوانی {len(refresh)} صفحه را پاک و دوباره می‌سازد: '
+            f'{chunks.count()} chunk، که {verified} تایشان رأیِ «agrees» دارند '
+            f'و رأیشان از بین می‌رود.'
+        ))
+        answer = input('ادامه بدهم؟ [yes/no] ').strip().lower()
+        return answer in ('y', 'yes', 'بله')
+
+    def _plan(self, book, collection, wanted, refresh=frozenset()):
         self.stdout.write(self.style.MIGRATE_HEADING(
             f'نقشه‌ی «{book.key}» → collection «{collection}»  (فایل: {book.filename})'
         ))
-        total_pages = total_left = 0
+        total_pages = total_left = total_refresh = 0
         for chapter in wanted:
             first, last = book.pdf_pages(chapter)
             pages = last - first + 1
             done = len(_done_pages(collection, chapter.title))
-            left = max(pages - done, 0)
+            again = sum(1 for p in refresh if first <= p <= last)
+            left = max(pages - done, 0) + again
             total_pages += pages
             total_left += left
+            total_refresh += again
             self.stdout.write(
                 f'  فصل {chapter.number}: {chapter.title}\n'
                 f'      صفحه‌ی کتاب {chapter.first_page}–{chapter.last_page}'
-                f'  →  صفحه‌ی PDF {first}–{last}   ({pages} صفحه، {done} انجام‌شده، {left} مانده)'
+                f'  →  صفحه‌ی PDF {first}–{last}   ({pages} صفحه، {done} انجام‌شده، {left} مانده'
+                + (f'، {again} بازخوانی)' if again else ')')
             )
         self.stdout.write('')
         self.stdout.write(f'جمع: {total_pages} صفحه، {total_left} صفحه مانده.')
+        if total_refresh:
+            chunks = LessonChunk.objects.filter(
+                collection=collection, source_page__in=refresh)
+            self.stdout.write(self.style.WARNING(
+                f'از آن، {total_refresh} صفحه بازخوانی است: {chunks.count()} chunkِ موجود '
+                f'پاک و از نو ساخته می‌شود ({chunks.filter(solution_verdict="agrees").count()} '
+                f'تایشان رأیِ «agrees» دارند و باید دوباره verify شوند).'
+            ))
         from django.conf import settings
 
         model = settings.OPENAI_VISION_MODEL
         per_page = estimate_page_cost(model)
         self.stdout.write(self.style.WARNING(
             f'هر صفحه یک فراخوانیِ vision دارد ({model}، ~${per_page:.4f} در صفحه، '
-            f'اندازه‌گیری‌شده روی ۱۱۱ صفحه‌ی واقعی): '
-            f'حدود ${total_left * per_page:.2f} برای صفحه‌های مانده '
-            f'(با Batch API نصف: ${total_left * per_page / 2:.2f}).'
+            f'از پنلِ صورتحساب اندازه‌گیری شده): '
+            f'حدود ${total_left * per_page:.2f} برای صفحه‌های مانده.'
         ))
 
     def _progress(self, event):
@@ -138,5 +235,9 @@ class Command(BaseCommand):
             self.stdout.write(self.style.ERROR(
                 f"    صفحه {event['page']}: خطا — {event['error'][:120]}"
             ))
+        elif kind == 'page_cleared':
+            self.stdout.write(self.style.WARNING(
+                f"    صفحه {event['page']}: {event['chunks_deleted']} chunkِ قدیمی پاک شد"
+            ))
         elif kind == 'budget_reached':
             self.stdout.write(self.style.WARNING('    سقفِ --max-pages پر شد؛ توقف.'))

rag/services/ingest.py

@@ -7,11 +7,13 @@ DB (LessonChunk، منبعِ حقیقت) + upsert به Qdrant (index).
 """
 import logging
 import os
+import re
 import tempfile
 from pathlib import Path
 
 from django.conf import settings
 from django.db import transaction
+from django.db.models import Max
 
 from ..models import IngestedPage, LessonChunk
 from .chunking import chunk_and_summarize
@@ -30,11 +32,26 @@ logger = logging.getLogger(__name__)
 MODEL_PRICE_PER_MTOK = {
     'gpt-4o': (2.50, 10.00),
     'gpt-4o-mini': (0.15, 0.60),
+    'gpt-4.1-mini': (0.40, 1.60),
+    'gemini-3.6-flash': (0.30, 2.50),
+}
+
+# هزینه‌ی واقعیِ هر صفحه — از پنلِ صورتحسابِ GapGPT اندازه‌گیری شده (۱۴۰۵/۰۶/۲۸،
+# سه صفحه‌ی حسابان یازدهم)، نه از جدولِ بالا. قیمتِ GapGPT با قیمتِ رسمیِ
+# OpenAI یکی نیست و gemini را مثلاً سه برابرِ gpt-4o حساب می‌کند، پس تخمینِ
+# مبتنی بر جدول برای انتخابِ مدل گمراه‌کننده بود.
+MEASURED_PAGE_COST_USD = {
+    'gpt-4o': 0.00766,
+    'gpt-4.1-mini': 0.00222,
+    'gemini-3.6-flash': 0.02135,
 }
 
 
 def estimate_page_cost(model: str) -> float:
-    """هزینه‌ی تقریبیِ یک صفحه بر پایه‌ی مصرفِ اندازه‌گیری‌شده‌ی واقعی."""
+    """هزینه‌ی تقریبیِ یک صفحه؛ اگر اندازه‌گیریِ واقعی داریم، همان."""
+    measured = MEASURED_PAGE_COST_USD.get(model)
+    if measured is not None:
+        return measured
     price_in, price_out = MODEL_PRICE_PER_MTOK.get(model, (2.50, 10.00))
     return 1425 / 1e6 * price_in + 400 / 1e6 * price_out
 
@@ -292,6 +309,68 @@ def _textbook_path(book: Textbook) -> Path:
     return root / book.filename
 
 
+# ارجاع به چیزی که در متن نیامده. «روبه‌رو»/«مقابل» هم هست چون کتاب‌های
+# ریاضی مرتب «شکلِ روبه‌رو» و «جدولِ مقابل» می‌نویسند بی‌آنکه واژه‌ی «شکل» را
+# در همان جمله بیاورند.
+FIGURE_REFERENCE = re.compile(r'شکل|جدول|نمودار|تصویر|روبه‌رو|مقابل')
+
+
+def pages_needing_refresh(collection: str, *, figures: bool = True,
+                          unjudgeable: bool = True) -> set[int]:
+    """
+    صفحه‌هایی که ارزشِ خواندنِ دوباره دارند.
+
+    figures     : صفحه‌ای که دستِ‌کم یک chunkش به شکل/جدولِ نوشته‌نشده ارجاع دارد.
+    unjudgeable : صفحه‌ای که دستِ‌کم یک chunkش در بازبینی «قابلِ قضاوت نیست» شده.
+
+    ⚠️ واحد «صفحه» است نه chunk، چون vision یک صفحه را یک‌جا می‌خواند. پس
+    بازخوانیِ یک صفحه chunkهای سالمِ همان صفحه را هم دور می‌ریزد و از نو
+    می‌سازد — نسخه‌ی تازه معمولاً بهتر است ولی `solution_verdict`ش پاک می‌شود
+    و باید دوباره verify شود. فرمان این عدد را قبل از خرج کردن نشان می‌دهد.
+    """
+    pages: set[int] = set()
+    fields = ['source_page', 'problem', 'solution', 'solution_verdict']
+    for chunk in (LessonChunk.objects
+                  .filter(collection=collection, source_page__isnull=False)
+                  .only(*fields)):
+        if unjudgeable and chunk.solution_verdict == 'unjudgeable':
+            pages.add(chunk.source_page)
+        elif figures and FIGURE_REFERENCE.search(
+            f"{chunk.problem or ''} {chunk.solution or ''}"
+        ):
+            pages.add(chunk.source_page)
+    return pages
+
+
+def clear_page(collection: str, page: int) -> dict:
+    """
+    پاک‌کردنِ کاملِ یک صفحه: نقاطِ Qdrant، ردیف‌های LessonChunk، و نشانه‌ی
+    IngestedPage — تا `_done_pages` دیگر آن را «انجام‌شده» نبیند.
+
+    ⚠️ ترتیب عمدی است: اول Qdrant، بعد دیتابیس. `qdrant_point_id` یک uuid4 است
+    که فقط در ردیفِ DB نگهداری می‌شود، پس اگر اول ردیف‌ها را پاک کنیم و بعد
+    حذفِ Qdrant شکست بخورد، شناسه‌ها از بین رفته‌اند و آن نقاط برای همیشه
+    بی‌صاحب می‌مانند — نه قابلِ پیدا کردن، نه قابلِ پاک کردن. با این ترتیب،
+    شکستِ Qdrant یعنی استثنا بالا می‌رود و DB دست‌نخورده می‌ماند.
+
+    این تابع لازم است چون ingestِ دوباره **جایگزین نمی‌کند**: هر ردیفِ تازه
+    uuid4ِ تازه می‌گیرد، پس بدونِ پاک‌سازی، خواندنِ دوباره‌ی یک صفحه نسخه‌ی
+    تکراری کنارِ نسخه‌ی قدیمی می‌گذارد.
+    """
+    chunks = list(
+        LessonChunk.objects
+        .filter(collection=collection, source_page=page)
+        .only('id', 'qdrant_point_id')
+    )
+    point_ids = [str(c.qdrant_point_id) for c in chunks]
+    if point_ids:
+        delete_points(collection, point_ids)
+    LessonChunk.objects.filter(pk__in=[c.pk for c in chunks]).delete()
+    marks, _ = IngestedPage.objects.filter(collection=collection, page=page).delete()
+    return {'page': page, 'chunks_deleted': len(chunks), 'points_deleted': len(point_ids),
+            'marks_deleted': marks}
+
+
 def _done_pages(collection: str, chapter_title: str) -> set[int]:
     """
     صفحه‌هایی که نباید دوباره پردازش شوند.
@@ -314,14 +393,22 @@ def _done_pages(collection: str, chapter_title: str) -> set[int]:
 
 def ingest_book(*, subject: str, grade: str, chapters=None, dpi: int = 200,
                 max_pages: int | None = None, resume: bool = True,
-                progress=None) -> dict:
+                refresh_pages=None, progress=None) -> dict:
     """
     یک کتابِ درسیِ کامل از data/textbooks/ را فصل‌به‌فصل ingest می‌کند.
 
-    chapters   : فهرستِ شماره‌ی فصل‌ها؛ None یعنی همه‌ی فصل‌های کتاب.
-    max_pages  : سقفِ تعدادِ صفحه‌ی پردازش‌شده در این اجرا (برای کنترلِ هزینه).
-    resume     : صفحه‌هایی که قبلاً chunk ساخته‌اند دوباره پردازش نشوند.
-    progress   : callback(dict) برای گزارشِ زنده به فرمانِ مدیریتی.
+    chapters     : فهرستِ شماره‌ی فصل‌ها؛ None یعنی همه‌ی فصل‌های کتاب.
+    max_pages    : سقفِ تعدادِ صفحه‌ی پردازش‌شده در این اجرا (برای کنترلِ هزینه).
+    resume       : صفحه‌هایی که قبلاً chunk ساخته‌اند دوباره پردازش نشوند.
+    refresh_pages: شماره‌ی صفحه‌های PDF که باید **بازخوانی** شوند — دادهٔ قبلیِ
+                   هر کدام درست پیش از خواندنِ دوباره‌اش پاک می‌شود. برای
+                   جبرانِ صفحاتی که پیش از قاعده‌ی self-contained خوانده شده‌اند.
+    progress     : callback(dict) برای گزارشِ زنده به فرمانِ مدیریتی.
+
+    ⚠️ پاک‌سازی عمداً صفحه‌به‌صفحه و بلافاصله قبل از پردازشِ همان صفحه انجام
+    می‌شود، نه یک‌جا در ابتدا: اگر اجرا وسطِ کار قطع شود (Ctrl-C، تمام شدنِ
+    اعتبار، سقفِ حساب)، فقط همان یک صفحه بدونِ داده می‌ماند و بقیه دست‌نخورده‌اند.
+    پاک‌سازیِ یک‌جا یعنی یک قطعیِ ساده، ۲۶۹ صفحه را خالی جا می‌گذارد.
 
     خروجی: خلاصه‌ی اجرا؛ هیچ استثنایی برای خطای یک صفحه بالا نمی‌آید — صفحه‌ی
     خطادار در `failed_pages` گزارش می‌شود و اجرای بعدی دوباره سراغش می‌رود.
@@ -357,20 +444,29 @@ def ingest_book(*, subject: str, grade: str, chapters=None, dpi: int = 200,
             progress(event)
 
     totals = {'pages_done': 0, 'pages_skipped': 0, 'chunks_created': 0,
-              'prompt_tokens': 0, 'completion_tokens': 0}
+              'chunks_replaced': 0, 'prompt_tokens': 0, 'completion_tokens': 0}
     failed_pages: list[int] = []
     budget = max_pages
 
+    refresh = {int(p) for p in (refresh_pages or ())}
+
     for chapter in wanted:
         first_pdf, last_pdf = book.pdf_pages(chapter)
         already = _done_pages(collection, chapter.title) if resume else set()
+        # صفحه‌ای که برای بازخوانی علامت خورده «انجام‌شده» حساب نمی‌شود، وگرنه
+        # منطقِ ازسرگیری ردش می‌کرد و بازخوانی بی‌سروصدا هیچ کاری نمی‌کرد.
+        already -= refresh
         emit(kind='chapter', chapter=chapter.number, title=chapter.title,
              pages=(first_pdf, last_pdf), already=len(already))
 
         # order در هر فصل ادامه‌دار می‌ماند تا اجرای دوم شماره‌ها را از نو نزند.
-        order = LessonChunk.objects.filter(
-            collection=collection, chapter_title=chapter.title,
-        ).count()
+        # از بیشینه گرفته می‌شود نه از تعداد: بازخوانی ردیف‌هایی را از وسط پاک
+        # می‌کند و آن‌وقت count() عددی می‌داد که با orderهای موجود تصادم داشت.
+        order = (
+            LessonChunk.objects
+            .filter(collection=collection, chapter_title=chapter.title)
+            .aggregate(m=Max('order'))['m'] or 0
+        )
 
         for page in range(first_pdf, last_pdf + 1):
             if page in already:
@@ -380,6 +476,29 @@ def ingest_book(*, subject: str, grade: str, chapters=None, dpi: int = 200,
                 emit(kind='budget_reached')
                 return _book_result(book, collection, totals, failed_pages, partial=True)
 
+            # پاک‌سازی فقط وقتی که واقعاً می‌خواهیم همین صفحه را بخوانیم —
+            # بعد از چکِ بودجه، تا رسیدن به سقف صفحه‌ای را خالی جا نگذارد.
+            if page in refresh:
+                try:
+                    removed = clear_page(collection, page)
+                except Exception as exc:  # noqa: BLE001
+                    # پاک‌سازی نشد یعنی خواندنِ دوباره نسخه‌ی تکراری می‌سازد؛
+                    # پس این صفحه را اصلاً نمی‌خوانیم و دست‌نخورده رد می‌شویم.
+                    logger.exception('rag.ingest_book: پاک‌سازیِ صفحه %s شکست خورد', page)
+                    failed_pages.append(page)
+                    emit(kind='page_failed', page=page,
+                         error=f'پاک‌سازی برای بازخوانی شکست خورد: {exc}')
+                    # بودجه اینجا هم کم می‌شود، مثلِ هر صفحه‌ی خطادارِ دیگر.
+                    # فراخوانیِ vision نرفته، ولی اگر Qdrant به‌کل قطع باشد
+                    # بدونِ این، حلقه کلِ فصل را بی‌ثمر طی می‌کند و
+                    # `--max-pages` — که کاربر برای مهارِ همین اجرا گذاشته —
+                    # هیچ اثری نمی‌گذارد.
+                    if budget is not None:
+                        budget -= 1
+                    continue
+                totals['chunks_replaced'] += removed['chunks_deleted']
+                emit(kind='page_cleared', **removed)
+
             try:
                 images = pdf_service.extract_page_images(
                     str(path), first_page=page, last_page=page, dpi=dpi,
@@ -472,6 +591,7 @@ def _book_result(book: Textbook, collection: str, totals: dict,
         'pages_processed': totals['pages_done'],
         'pages_skipped': totals['pages_skipped'],
         'chunks_created': totals['chunks_created'],
+        'chunks_replaced': totals.get('chunks_replaced', 0),
         'failed_pages': failed_pages,
         'prompt_tokens': totals['prompt_tokens'],
         'completion_tokens': totals['completion_tokens'],