perf: batch body fetches when many new UIDs are pending

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

docs/superpowers/plans/2026-07-20-mail-fetch-fix.md

@@ -0,0 +1,662 @@
+# Mail Fetch Re-download Fix Implementation Plan
+
+> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
+
+**Goal:** Stop `paperless_mail` from re-downloading full IMAP message bodies for mail it has already handled, and make sure attachment-less mail under attachments-only rules is recorded so it stops matching the search forever.
+
+**Architecture:** In `_handle_mail_rule`, replace the single "fetch every matching message's full body" call with a cheap `UID SEARCH` first, subtract UIDs already in `ProcessedMail`, and fetch bodies only for the remainder (batched). Separately, extract the existing "record `PROCESSED_WO_CONSUMPTION`" block into a shared helper and call it from the no-attachments-under-attachments-only-scope early return in `_handle_message`, which currently skips it entirely.
+
+**Tech Stack:** Django (pytest-django), `imap_tools` (`MailBox.uids()`, `AND` query builder), existing `ProcessedMail` model — no new dependencies, no migrations.
+
+## Global Constraints
+
+- No new required settings and no DB migrations for existing installs (spec: "No new required settings/migrations for existing installs").
+- Batch size for body fetches is a hardcoded module constant (`MAIL_FETCH_BATCH_SIZE = 500`), not user-configurable.
+- No mailbox action (mark-read/flag/tag/move/delete) is applied for the no-attachment/no-consumption case — only a `ProcessedMail` row is written, matching the existing precedent for other no-consumption outcomes.
+- Every path that currently produces a document or applies a mail action must behave exactly as before (no change to `_process_attachments`'/`_process_eml`'s consuming behavior, action application, or the existing per-message `already_processed` dedup check, which stays in place as a safety net).
+- Follow the existing per-message `uid_validity` matching semantics exactly: when `self._current_uid_validity is not None`, only match `ProcessedMail` rows with the same `uid_validity` or `uid_validity IS NULL`; when it is `None` (server didn't report UIDVALIDITY), fall back to matching on `(rule, uid, folder)` alone, ignoring `uid_validity`. (This exact conditional already exists at `mail.py:722-726` for the per-message check — the new UID-diff query must replicate it or several existing uidvalidity tests will break.)
+
+---
+
+### Task 1: Add UID search support to the `BogusMailBox` test double
+
+**Files:**
+
+- Modify: `src/paperless_mail/tests/test_mail.py:137-171` (`BogusMailBox.fetch`)
+- Test: `src/paperless_mail/tests/test_mail.py` (new test in `class TestMail`)
+
+**Interfaces:**
+
+- Produces: `BogusMailBox.uids(self, criteria, charset="") -> list[str]`, filtering with the same rules as `fetch()`, returning only UIDs (no message bodies). `BogusMailBox.fetch()` gains support for a `UID <comma-list>` criteria token (used by real code's `AND(uid=[...])` queries).
+- Consumes: nothing new — this is test infrastructure only, no production code changes in this task.
+
+- [ ] **Step 1: Write the failing test**
+
+Add to `class TestMail` in `src/paperless_mail/tests/test_mail.py` (near the other `BogusMailBox`-adjacent tests, e.g. after `test_handle_message`):
+
+```python
+    def test_bogus_mailbox_uids_and_uid_criteria(self) -> None:
+        mailbox = self.mailMocker.bogus_mailbox
+        all_messages = list(mailbox.messages)
+
+        # uids() returns the UIDs of unseen messages, no bodies needed to call it
+        unseen_uids = mailbox.uids("(UNSEEN)")
+        self.assertEqual(
+            set(unseen_uids),
+            {m.uid for m in all_messages if not m.seen},
+        )
+
+        # fetch() with an explicit UID criteria returns only the matching messages
+        target_uid = all_messages[0].uid
+        from imap_tools import AND
+
+        fetched = mailbox.fetch(AND(uid=[target_uid]), mark_seen=False)
+        self.assertEqual([m.uid for m in fetched], [target_uid])
+```
+
+- [ ] **Step 2: Run test to verify it fails**
+
+Run: `uv run pytest --override-ini="addopts=" src/paperless_mail/tests/test_mail.py::TestMail::test_bogus_mailbox_uids_and_uid_criteria -v`
+Expected: FAIL with `AttributeError: 'BogusMailBox' object has no attribute 'uids'` (and/or the UID-criteria fetch returning all 3 default messages instead of 1, since `fetch()` doesn't understand a `UID` token yet).
+
+- [ ] **Step 3: Implement `uids()` and `UID` criteria support**
+
+Replace `src/paperless_mail/tests/test_mail.py:137-171`:
+
+```python
+    def fetch(self, criteria, mark_seen, charset="", *, bulk=True):
+        return self._filter_messages(criteria)
+
+    def uids(self, criteria, charset="") -> list[str]:
+        return [m.uid for m in self._filter_messages(criteria)]
+
+    def _filter_messages(self, criteria):
+        msg = self.messages
+
+        criteria = str(criteria).strip("()").split(" ")
+
+        if "UNSEEN" in criteria:
+            msg = filter(lambda m: not m.seen, msg)
+
+        if "SUBJECT" in criteria:
+            subject = criteria[criteria.index("SUBJECT") + 1].strip('"')
+            msg = filter(lambda m: subject in m.subject, msg)
+
+        if "BODY" in criteria:
+            body = criteria[criteria.index("BODY") + 1].strip('"')
+            msg = filter(lambda m: body in m.text, msg)
+
+        if "FROM" in criteria:
+            from_ = criteria[criteria.index("FROM") + 1].strip('"')
+            msg = filter(lambda m: from_ in m.from_, msg)
+
+        if "TO" in criteria:
+            to_ = criteria[criteria.index("TO") + 1].strip('"')
+            msg = filter(lambda m: any(to_ in to_addr for to_addr in m.to), msg)
+
+        if "UNFLAGGED" in criteria:
+            msg = filter(lambda m: not m.flagged, msg)
+
+        if "UNKEYWORD" in criteria:
+            tag = criteria[criteria.index("UNKEYWORD") + 1].strip("'")
+            msg = filter(lambda m: tag not in m.flags, msg)
+
+        if "(X-GM-LABELS" in criteria:  # ['NOT', '(X-GM-LABELS', '"processed"']
+            msg = filter(lambda m: "processed" not in m.flags, msg)
+
+        if "UID" in criteria:
+            uid_list = criteria[criteria.index("UID") + 1].split(",")
+            msg = filter(lambda m: m.uid in uid_list, msg)
+
+        return list(msg)
+```
+
+This is a pure refactor of the existing filtering logic into `_filter_messages`, reused by both `fetch()` (unchanged behavior) and the new `uids()`, plus one new `if "UID" in criteria` branch.
+
+- [ ] **Step 4: Run test to verify it passes**
+
+Run: `uv run pytest --override-ini="addopts=" src/paperless_mail/tests/test_mail.py::TestMail::test_bogus_mailbox_uids_and_uid_criteria -v`
+Expected: PASS
+
+- [ ] **Step 5: Run the full existing `test_mail.py` suite to confirm the refactor didn't change `fetch()` behavior**
+
+Run: `uv run pytest --override-ini="addopts=" src/paperless_mail/tests/test_mail.py -v`
+Expected: all tests PASS (same pass count as before this task)
+
+- [ ] **Step 6: Commit**
+
+```bash
+git add src/paperless_mail/tests/test_mail.py
+git commit -m "test: add BogusMailBox.uids() and UID criteria support"
+```
+
+---
+
+### Task 2: Record `ProcessedMail` for attachment-less mail under attachments-only rules
+
+**Files:**
+
+- Modify: `src/paperless_mail/mail.py:749-793` (`_handle_message`)
+- Modify: `src/paperless_mail/mail.py:952-979` (`_process_attachments`, tail)
+- Test: `src/paperless_mail/tests/test_mail.py:540-548` (`test_handle_empty_message`, rewritten)
+
+**Interfaces:**
+
+- Produces: `MailAccountHandler._record_processed_without_consumption(self, message: MailMessage, rule: MailRule) -> None` — idempotently writes a `ProcessedMail(status="PROCESSED_WO_CONSUMPTION")` row for `(rule, message.uid, rule.folder)` if one doesn't already exist for the current `self._current_uid_validity`. No mailbox action is applied.
+- Consumes: `self._current_uid_validity` (already set by `_handle_mail_rule` before `_handle_message` is called; `None` when called directly, e.g. in unit tests).
+
+- [ ] **Step 1: Write the failing test**
+
+Replace `src/paperless_mail/tests/test_mail.py:540-548` (`test_handle_empty_message`):
+
+```python
+    def test_handle_empty_message(self) -> None:
+        message = self.mailMocker.messageBuilder.create_message(
+            subject="No attachments here",
+            attachments=[],
+        )
+
+        account = MailAccount.objects.create()
+        rule = MailRule.objects.create(
+            account=account,
+            consumption_scope=MailRule.ConsumptionScope.ATTACHMENTS_ONLY,
+        )
+
+        result = self.mail_account_handler._handle_message(message, rule)
+
+        self.mailMocker._queue_consumption_tasks_mock.assert_not_called()
+        self.assertEqual(result, 0)
+
+        processed = ProcessedMail.objects.get(
+            rule=rule,
+            uid=message.uid,
+            folder=rule.folder,
+        )
+        self.assertEqual(processed.status, "PROCESSED_WO_CONSUMPTION")
+
+        # Calling it again must not create a second row
+        self.mail_account_handler._handle_message(message, rule)
+        self.assertEqual(
+            ProcessedMail.objects.filter(
+                rule=rule,
+                uid=message.uid,
+                folder=rule.folder,
+            ).count(),
+            1,
+        )
+```
+
+- [ ] **Step 2: Run test to verify it fails**
+
+Run: `uv run pytest --override-ini="addopts=" src/paperless_mail/tests/test_mail.py::TestMail::test_handle_empty_message -v`
+Expected: FAIL — `ProcessedMail.DoesNotExist` (no row is created by the current early return).
+
+- [ ] **Step 3: Add the shared helper and call it from both places**
+
+Insert a new method right after `_handle_message` ends, i.e. after `src/paperless_mail/mail.py:793` (`return processed_elements`), before `def filename_inclusion_matches`:
+
+```python
+    def _record_processed_without_consumption(
+        self,
+        message: MailMessage,
+        rule: MailRule,
+    ) -> None:
+        if not ProcessedMail.objects.filter(
+            rule=rule,
+            uid=message.uid,
+            folder=rule.folder,
+            uid_validity=self._current_uid_validity,
+        ).exists():
+            ProcessedMail.objects.create(
+                rule=rule,
+                folder=rule.folder,
+                uid=message.uid,
+                uid_validity=self._current_uid_validity,
+                subject=message.subject,
+                received=make_aware(message.date)
+                if is_naive(message.date)
+                else message.date,
+                status="PROCESSED_WO_CONSUMPTION",
+            )
+```
+
+Then modify `_handle_message`'s early return at `src/paperless_mail/mail.py:756-760`:
+
+```python
+        # Skip Message handling when only attachments are to be processed but
+        # message doesn't have any.
+        if (
+            not message.attachments
+            and rule.consumption_scope == MailRule.ConsumptionScope.ATTACHMENTS_ONLY
+        ):
+            return processed_elements
+```
+
+becomes:
+
+```python
+        # Skip Message handling when only attachments are to be processed but
+        # message doesn't have any.
+        if (
+            not message.attachments
+            and rule.consumption_scope == MailRule.ConsumptionScope.ATTACHMENTS_ONLY
+        ):
+            self._record_processed_without_consumption(message, rule)
+            return processed_elements
+```
+
+Then replace the tail of `_process_attachments` at `src/paperless_mail/mail.py:952-979`:
+
+```python
+        if len(consume_tasks) > 0:
+            queue_consumption_tasks(
+                consume_tasks=consume_tasks,
+                rule=rule,
+                message=message,
+                uid_validity=self._current_uid_validity,
+            )
+        else:
+            # No files to consume, just mark as processed if it wasn't by .eml processing
+            if not ProcessedMail.objects.filter(
+                rule=rule,
+                uid=message.uid,
+                folder=rule.folder,
+                uid_validity=self._current_uid_validity,
+            ).exists():
+                ProcessedMail.objects.create(
+                    rule=rule,
+                    folder=rule.folder,
+                    uid=message.uid,
+                    uid_validity=self._current_uid_validity,
+                    subject=message.subject,
+                    received=make_aware(message.date)
+                    if is_naive(message.date)
+                    else message.date,
+                    status="PROCESSED_WO_CONSUMPTION",
+                )
+
+        return processed_attachments
+```
+
+with:
+
+```python
+        if len(consume_tasks) > 0:
+            queue_consumption_tasks(
+                consume_tasks=consume_tasks,
+                rule=rule,
+                message=message,
+                uid_validity=self._current_uid_validity,
+            )
+        else:
+            # No files to consume, just mark as processed if it wasn't by .eml processing
+            self._record_processed_without_consumption(message, rule)
+
+        return processed_attachments
+```
+
+- [ ] **Step 4: Run test to verify it passes**
+
+Run: `uv run pytest --override-ini="addopts=" src/paperless_mail/tests/test_mail.py::TestMail::test_handle_empty_message -v`
+Expected: PASS
+
+- [ ] **Step 5: Run the full `test_mail.py` suite (regression check for the `_process_attachments` refactor)**
+
+Run: `uv run pytest --override-ini="addopts=" src/paperless_mail/tests/test_mail.py -v`
+Expected: all PASS
+
+- [ ] **Step 6: Commit**
+
+```bash
+git add src/paperless_mail/mail.py src/paperless_mail/tests/test_mail.py
+git commit -m "fix: record ProcessedMail for attachment-less mail under attachments-only rules"
+```
+
+---
+
+### Task 3: Diff UIDs against `ProcessedMail` before fetching message bodies
+
+**Files:**
+
+- Modify: `src/paperless_mail/mail.py:641-693` (`_handle_mail_rule`, the fetch section)
+- Test: `src/paperless_mail/tests/test_mail.py` (new test in `class TestMail`)
+
+**Interfaces:**
+
+- Consumes: `MailAccountHandler._record_processed_without_consumption` (Task 2), `ProcessedMail` model, `imap_tools.AND`, `M.uids()` (Task 1's `BogusMailBox.uids()` in tests; `imap_tools.MailBox.uids()` in production).
+- Produces: no new public interface — `_handle_mail_rule`'s external behavior (return value, exceptions raised) is unchanged; only its internal fetch strategy changes. `_handle_mail_rule` now returns `0` immediately, without calling `M.fetch()` at all, when every UID matching the search criteria already has a `ProcessedMail` row.
+
+- [ ] **Step 1: Write the failing test**
+
+Add to `class TestMail` in `src/paperless_mail/tests/test_mail.py`:
+
+```python
+    def test_handle_mail_account_skips_body_fetch_for_already_processed_mail(
+        self,
+    ) -> None:
+        """
+        GIVEN:
+            - An attachment-less mail under an attachments-only mark-read rule,
+              already recorded as PROCESSED_WO_CONSUMPTION
+        WHEN:
+            - The mail account is processed again and the mail still matches the
+              search criteria (it was never marked read, since no mail action is
+              applied for the no-consumption case)
+        THEN:
+            - No IMAP body fetch happens for that mail; only the cheap UID search runs.
+        """
+        account = MailAccount.objects.create(
+            name="test",
+            imap_server="",
+            username="admin",
+            password="secret",
+        )
+        rule = MailRule.objects.create(
+            name="testrule",
+            account=account,
+            action=MailRule.MailAction.MARK_READ,
+            consumption_scope=MailRule.ConsumptionScope.ATTACHMENTS_ONLY,
+        )
+
+        message = self.mailMocker.messageBuilder.create_message(
+            subject="No attachment",
+            attachments=[],
+        )
+        self.mailMocker.bogus_mailbox.messages = [message]
+        self.mailMocker.bogus_mailbox.updateClient()
+
+        # First run: records ProcessedMail without consuming anything.
+        self.mail_account_handler.handle_mail_account(account)
+        self.assertTrue(
+            ProcessedMail.objects.filter(
+                rule=rule,
+                uid=message.uid,
+                folder=rule.folder,
+            ).exists(),
+        )
+        self.mailMocker._queue_consumption_tasks_mock.assert_not_called()
+
+        # Second run: message still matches UNSEEN (mark-read action never ran),
+        # but its body must not be downloaded again.
+        with mock.patch.object(
+            self.mailMocker.bogus_mailbox,
+            "fetch",
+            wraps=self.mailMocker.bogus_mailbox.fetch,
+        ) as fetch_spy:
+            self.mail_account_handler.handle_mail_account(account)
+
+        fetch_spy.assert_not_called()
+```
+
+- [ ] **Step 2: Run test to verify it fails**
+
+Run: `uv run pytest --override-ini="addopts=" src/paperless_mail/tests/test_mail.py::TestMail::test_handle_mail_account_skips_body_fetch_for_already_processed_mail -v`
+Expected: FAIL — `fetch_spy.assert_not_called()` fails because the current code always calls `M.fetch()` for every message matching the search criteria, regardless of `ProcessedMail`.
+
+- [ ] **Step 3: Implement the UID diff**
+
+Replace `src/paperless_mail/mail.py:683-693`:
+
+```python
+        try:
+            messages = M.fetch(
+                criteria=criterias,
+                mark_seen=False,
+                charset=rule.account.character_set,
+                bulk=True,
+            )
+        except Exception as err:
+            raise MailError(
+                f"Rule {rule}: Error while fetching folder {rule.folder}",
+            ) from err
+```
+
+with:
+
+```python
+        try:
+            all_uids = set(
+                M.uids(criteria=criterias, charset=rule.account.character_set),
+            )
+        except Exception as err:
+            raise MailError(
+                f"Rule {rule}: Error while searching folder {rule.folder}",
+            ) from err
+
+        processed_uids_qs = ProcessedMail.objects.filter(
+            rule=rule,
+            folder=rule.folder,
+            uid__in=all_uids,
+        )
+        if self._current_uid_validity is not None:
+            processed_uids_qs = processed_uids_qs.filter(
+                Q(uid_validity=self._current_uid_validity)
+                | Q(uid_validity__isnull=True),
+            )
+        processed_uids = set(processed_uids_qs.values_list("uid", flat=True))
+
+        new_uids = all_uids - processed_uids
+
+        if not new_uids:
+            self.log.debug(
+                f"Rule {rule}: No new mail matching criteria {criterias}",
+            )
+            return 0
+
+        try:
+            messages = M.fetch(
+                criteria=AND(uid=list(new_uids)),
+                mark_seen=False,
+                charset=rule.account.character_set,
+                bulk=True,
+            )
+        except Exception as err:
+            raise MailError(
+                f"Rule {rule}: Error while fetching folder {rule.folder}",
+            ) from err
+```
+
+Note the `uid_validity` handling here deliberately mirrors the existing per-message check at `mail.py:722-726` exactly (see Global Constraints) — the extra `Q(...)` filter is applied only when `self._current_uid_validity is not None`.
+
+- [ ] **Step 4: Run test to verify it passes**
+
+Run: `uv run pytest --override-ini="addopts=" src/paperless_mail/tests/test_mail.py::TestMail::test_handle_mail_account_skips_body_fetch_for_already_processed_mail -v`
+Expected: PASS
+
+- [ ] **Step 5: Run the full `test_mail.py` suite (regression check, especially the uidvalidity tests)**
+
+Run: `uv run pytest --override-ini="addopts=" src/paperless_mail/tests/test_mail.py -v`
+Expected: all PASS, including:
+
+- `test_handle_mail_account_skip_duplicate_uids_from_fetch`
+- `test_handle_mail_account_skips_mail_already_processed_in_same_uidvalidity`
+- `test_handle_mail_account_processes_mail_after_uidvalidity_change`
+- `test_handle_mail_account_skips_mail_processed_before_uidvalidity_tracking`
+- `test_handle_mail_account_processes_mail_when_uidvalidity_unavailable`
+- `test_handle_mail_account_skips_mail_when_uidvalidity_unavailable_but_prior_record_exists`
+- `test_handle_mail_account_overlapping_rules_only_first_consumes`
+
+If any of these fail, the `uid_validity` branching in Step 3 does not match `mail.py:722-726` closely enough — re-check against the Global Constraints note before changing test expectations.
+
+- [ ] **Step 6: Commit**
+
+```bash
+git add src/paperless_mail/mail.py src/paperless_mail/tests/test_mail.py
+git commit -m "fix: diff UIDs against ProcessedMail before fetching message bodies"
+```
+
+---
+
+### Task 4: Batch the body fetch for large backlogs
+
+**Files:**
+
+- Modify: `src/paperless_mail/mail.py:74-76` (module-level constant)
+- Modify: `src/paperless_mail/mail.py` (the `M.fetch(...)` block added in Task 3)
+- Test: `src/paperless_mail/tests/test_mail.py` (new test in `class TestMail`)
+
+**Interfaces:**
+
+- Produces: module-level constant `paperless_mail.mail.MAIL_FETCH_BATCH_SIZE: int = 500`.
+- Consumes: `itertools` (already imported at `mail.py:2`), `sorted_new_uids` derived from Task 3's `new_uids` set.
+
+- [ ] **Step 1: Write the failing test**
+
+Add to `class TestMail` in `src/paperless_mail/tests/test_mail.py`. Add `from paperless_mail.mail import MAIL_FETCH_BATCH_SIZE` to the imports at the top of the file (alongside the other `from paperless_mail.mail import ...` lines at `test_mail.py:35-38`).
+
+```python
+    @mock.patch("paperless_mail.mail.MAIL_FETCH_BATCH_SIZE", 5)
+    def test_handle_mail_account_batches_body_fetch_for_large_backlog(self) -> None:
+        """
+        GIVEN:
+            - More new/unprocessed mail than MAIL_FETCH_BATCH_SIZE
+        WHEN:
+            - The mail account is processed
+        THEN:
+            - The body fetch is issued in multiple batches
+            - Every message is still processed (none dropped at a batch boundary)
+        """
+        account = MailAccount.objects.create(
+            name="test",
+            imap_server="",
+            username="admin",
+            password="secret",
+        )
+        rule = MailRule.objects.create(
+            name="testrule",
+            account=account,
+            action=MailRule.MailAction.MARK_READ,
+            consumption_scope=MailRule.ConsumptionScope.ATTACHMENTS_ONLY,
+        )
+
+        message_count = 12  # more than the patched batch size of 5
+        self.mailMocker.bogus_mailbox.messages = [
+            self.mailMocker.messageBuilder.create_message(
+                subject=f"No attachment {i}",
+                attachments=[],
+            )
+            for i in range(message_count)
+        ]
+        self.mailMocker.bogus_mailbox.updateClient()
+
+        with mock.patch.object(
+            self.mailMocker.bogus_mailbox,
+            "fetch",
+            wraps=self.mailMocker.bogus_mailbox.fetch,
+        ) as fetch_spy:
+            self.mail_account_handler.handle_mail_account(account)
+
+        # ceil(12 / 5) == 3 batches
+        self.assertEqual(fetch_spy.call_count, 3)
+        self.assertEqual(
+            ProcessedMail.objects.filter(rule=rule).count(),
+            message_count,
+        )
+```
+
+- [ ] **Step 2: Run test to verify it fails**
+
+Run: `uv run pytest --override-ini="addopts=" src/paperless_mail/tests/test_mail.py::TestMail::test_handle_mail_account_batches_body_fetch_for_large_backlog -v`
+Expected: FAIL — either an `ImportError` for `MAIL_FETCH_BATCH_SIZE` (doesn't exist yet) or, once that import is stubbed out, `fetch_spy.call_count == 1` instead of `3` (Task 3's implementation fetches all new UIDs in one call).
+
+- [ ] **Step 3: Add the constant and batch the fetch**
+
+Insert after `src/paperless_mail/mail.py:74` (right after the `APPLE_MAIL_TAG_COLORS` dict closes, before `class MailError`):
+
+```python
+MAIL_FETCH_BATCH_SIZE = 500
+```
+
+Replace the `M.fetch(...)` block added in Task 3 (Task 3 Step 3's final `try/except`):
+
+```python
+        try:
+            messages = M.fetch(
+                criteria=AND(uid=list(new_uids)),
+                mark_seen=False,
+                charset=rule.account.character_set,
+                bulk=True,
+            )
+        except Exception as err:
+            raise MailError(
+                f"Rule {rule}: Error while fetching folder {rule.folder}",
+            ) from err
+```
+
+with:
+
+```python
+        sorted_new_uids = sorted(new_uids, key=int)
+        message_batches = []
+        for batch_start in range(0, len(sorted_new_uids), MAIL_FETCH_BATCH_SIZE):
+            batch = sorted_new_uids[batch_start : batch_start + MAIL_FETCH_BATCH_SIZE]
+            try:
+                message_batches.append(
+                    M.fetch(
+                        criteria=AND(uid=batch),
+                        mark_seen=False,
+                        charset=rule.account.character_set,
+                        bulk=True,
+                    ),
+                )
+            except Exception as err:
+                raise MailError(
+                    f"Rule {rule}: Error while fetching folder {rule.folder}",
+                ) from err
+
+        messages = itertools.chain(*message_batches)
+```
+
+- [ ] **Step 4: Run test to verify it passes**
+
+Run: `uv run pytest --override-ini="addopts=" src/paperless_mail/tests/test_mail.py::TestMail::test_handle_mail_account_batches_body_fetch_for_large_backlog -v`
+Expected: PASS
+
+- [ ] **Step 5: Run the full `test_mail.py` suite**
+
+Run: `uv run pytest --override-ini="addopts=" src/paperless_mail/tests/test_mail.py -v`
+Expected: all PASS
+
+- [ ] **Step 6: Commit**
+
+```bash
+git add src/paperless_mail/mail.py src/paperless_mail/tests/test_mail.py
+git commit -m "perf: batch body fetches when many new UIDs are pending"
+```
+
+---
+
+### Task 5: Full regression pass and lint
+
+**Files:** none (verification only, plus any fixups this step surfaces)
+
+- [ ] **Step 1: Run the full backend test suite with coverage/parallelism as configured**
+
+Run: `uv run pytest src/paperless_mail/`
+Expected: all `paperless_mail` tests PASS
+
+- [ ] **Step 2: Run the full project test suite**
+
+Run: `uv run pytest`
+Expected: all tests PASS (no regressions outside `paperless_mail`)
+
+- [ ] **Step 3: Lint and format**
+
+Run: `uv run ruff check src/paperless_mail/` and `uv run ruff format --check src/paperless_mail/`
+Expected: no errors. If `ruff format --check` reports files needing formatting, run `uv run ruff format src/paperless_mail/` and re-check.
+
+- [ ] **Step 4: Type check against the frozen baselines**
+
+Run: `uv run mypy src/paperless_mail/mail.py` (or the project's standard mypy invocation) and confirm no new violations beyond `.mypy-baseline.txt`.
+Expected: no new errors introduced by this change.
+
+- [ ] **Step 5: Commit any fixups**
+
+If Steps 1-4 required changes:
+
+```bash
+git add -u
+git commit -m "chore: fix lint/type fallout from mail fetch fix"
+```
+
+If no changes were needed, skip this step — nothing to commit.

docs/superpowers/specs/2026-07-20-mail-fetch-fix-design.md

@@ -0,0 +1,216 @@
+# Mail fetch fix: avoid re-downloading already-handled IMAP mail
+
+## Problem
+
+`paperless_mail`'s `_handle_mail_rule` (`src/paperless_mail/mail.py:641`) fetches
+the full RFC822 body of every message an IMAP `SEARCH` matches (via
+`M.fetch(criteria=criterias, ..., bulk=True)`), then does de-duplication in
+Python afterward. Two things compound into a real bug (reported upstream in
+[paperless-ngx#13175](https://github.com/paperless-ngx/paperless-ngx/issues/13175)):
+
+1. **Full-body fetch happens before dedup.** The de-dup check against
+   `ProcessedMail` (`mail.py:717-731`) happens per-message, after the body has
+   already been downloaded for every matching message. Steady-state cost is
+   _O(everything the search matches)_, not _O(new since last run)_.
+2. **Attachment-less mail under an attachments-only rule is never recorded.**
+   `_handle_message` (`mail.py:749`) returns early, before any attachment
+   processing, when a message has no attachments and
+   `rule.consumption_scope == ATTACHMENTS_ONLY`:
+
+   ```python
+   if (
+       not message.attachments
+       and rule.consumption_scope == MailRule.ConsumptionScope.ATTACHMENTS_ONLY
+   ):
+       return processed_elements  # 0, no ProcessedMail row, no mail action applied
+   ```
+
+   No `ProcessedMail` row is written and no mail action (mark-read/flag/tag)
+   is applied, since both only happen via `queue_consumption_tasks`, which is
+   only reached from the attachment/eml processing paths. The message keeps
+   matching the search (e.g. `UNSEEN`) forever.
+
+Combined: an inbox where most mail has no attachments gets fully re-downloaded
+on every scheduled run (default every 10 minutes), indefinitely, for large
+mailboxes -- confirmed against the current codebase, not user
+misconfiguration.
+
+## Goals
+
+- Stop re-downloading full message bodies for mail that has already been
+  handled (processed into a document, or determined to produce nothing).
+- Record `ProcessedMail` for the no-attachment / attachments-only early-return
+  case, so it participates in dedup like every other terminal outcome.
+- No new required settings or DB migrations for existing installs.
+- Graceful behavior on first run against a large existing mailbox (no single
+  giant IMAP command, no giant in-memory batch).
+- Preserve existing action/consumption semantics exactly for every path that
+  currently produces a document or applies a mail action.
+
+## Non-goals
+
+- Changing what mail actions (mark-read/flag/tag/move/delete) get applied, or
+  when.
+- Applying a mail action to the no-attachment early-return case (explicitly
+  out of scope -- see Decisions).
+- Making batch size user-configurable.
+- Touching EML_ONLY / EVERYTHING consumption scopes' semantics (they already
+  don't hit the early-return branch described above).
+
+## Decisions
+
+These were settled during brainstorming and are load-bearing for the design
+below:
+
+- **Fix both halves together.** Recording `ProcessedMail` alone does not fix
+  the bandwidth problem: without a mail action applied, an attachment-less
+  message stays unseen and keeps matching the search criteria, so its body
+  gets re-downloaded every run regardless of whether it gets reprocessed.
+  Fixing only the bandwidth side without recording `ProcessedMail` would mean
+  the UID-diff step never has anything to exclude for these messages. They
+  must ship together.
+- **No mailbox action for the no-attachment case.** Only a `ProcessedMail`
+  row is written; the message is not marked read/flagged/moved/deleted. This
+  matches the existing precedent for other no-consumption outcomes
+  (`mail.py:960-977`, e.g. an attachment present but filtered out or an
+  unsupported mime type) and avoids changing the user's mailbox state for
+  mail paperless previously never touched.
+- **UID-diff-before-fetch, not a BODYSTRUCTURE probe or SEARCH-side
+  exclusion.** Considered and rejected:
+  - BODYSTRUCTURE probing (fetch structure only, decide whether to fetch full
+    body) only helps the attachments-only scope, not the general
+    already-processed case, and `imap_tools` doesn't cleanly expose a
+    structure-only fetch.
+  - Excluding already-processed UIDs directly in the IMAP `SEARCH` criteria
+    (`NOT UID (...)`) was rejected because for tens of thousands of
+    already-processed UIDs the excluded-UID list itself blows up the command
+    size -- worse than the two-step approach.
+- **Batch size is a hardcoded constant**, not a new setting, per the
+  "no new required settings" goal.
+
+## Design
+
+### `_handle_mail_rule` (`mail.py:641`)
+
+Replace the single `M.fetch(criteria=criterias, ...)` call with:
+
+1. `self._current_uid_validity` computed as today (unchanged, already first).
+2. `criterias = make_criterias(...)` (unchanged).
+3. `all_uids = set(M.uids(criteria=criterias, charset=rule.account.character_set))`
+   -- a `UID SEARCH`, no bodies.
+4. Query already-processed UIDs in one DB round trip, reusing the same
+   uid_validity matching already used per-message:
+
+   ```python
+   processed_uids = set(
+       ProcessedMail.objects.filter(
+           rule=rule, folder=rule.folder, uid__in=all_uids,
+       ).filter(
+           Q(uid_validity=self._current_uid_validity) | Q(uid_validity__isnull=True),
+       ).values_list("uid", flat=True)
+   )
+   ```
+
+5. `new_uids = all_uids - processed_uids`. If empty: log at debug level and
+   `return 0` -- no body fetch at all. This is the steady-state case.
+6. Otherwise, iterate `new_uids` in fixed-size batches of
+   `MAIL_FETCH_BATCH_SIZE = 500` (module-level constant), calling
+   `M.fetch(criteria=AND(uid=batch), mark_seen=False, charset=rule.account.character_set, bulk=True)`
+   per batch. Chain the resulting message iterators into the existing
+   per-message loop (`mail.py:699+`) unchanged.
+7. The existing per-message `already_processed` DB check
+   (`mail.py:717-731`) stays in place unchanged, as a safety net against
+   races (e.g. concurrent rule runs against the same folder in this pass) and
+   to keep behavior identical if steps 3/4 ever disagree with it.
+
+### New helper: `_record_processed_without_consumption`
+
+Extract the existing dedup-and-create block at `mail.py:960-977` into a
+method on the same class:
+
+```python
+def _record_processed_without_consumption(self, message, rule) -> None:
+    if not ProcessedMail.objects.filter(
+        rule=rule,
+        uid=message.uid,
+        folder=rule.folder,
+        uid_validity=self._current_uid_validity,
+    ).exists():
+        ProcessedMail.objects.create(
+            rule=rule,
+            folder=rule.folder,
+            uid=message.uid,
+            uid_validity=self._current_uid_validity,
+            subject=message.subject,
+            received=make_aware(message.date) if is_naive(message.date) else message.date,
+            status="PROCESSED_WO_CONSUMPTION",
+        )
+```
+
+Call sites:
+
+- `_process_attachments`'s existing no-consumables branch (replaces the
+  inline block, same behavior).
+- `_handle_message`'s early return (`mail.py:756-760`), newly, for the
+  no-attachments-under-attachments-only-scope case.
+
+### Data flow (steady state, nothing new)
+
+search criteria (unchanged) -> `UID SEARCH` for matching UIDs (cheap) ->
+subtract UIDs already in `ProcessedMail` (cheap, DB-only) -> zero new UIDs ->
+no fetch, no body download.
+
+### Data flow (new mail present)
+
+... same as above -> some UIDs remain -> fetch bodies only for those,
+batched -> existing per-message processing, action application, and
+`ProcessedMail` recording, unchanged except the no-attachment early return
+now also calls `_record_processed_without_consumption`.
+
+## Error handling
+
+- `M.uids(...)` raising is wrapped in the same `try/except` that currently
+  wraps `M.fetch(...)` (`mail.py:683-693`), surfacing as `MailError`
+  identically to today's search-failure behavior.
+- A batch's `M.fetch(...)` raising mid-loop is caught by the same
+  `try/except`, applied per-batch. One failing batch fails the whole rule
+  run for this pass -- matching today's all-or-nothing semantics (currently
+  a single failed fetch already fails the whole rule).
+- Empty `new_uids` short-circuits before any fetch, so it can't hit fetch
+  error paths at all for the common case.
+- `_record_processed_without_consumption` reuses the existing
+  `ProcessedMail.objects.create` call already in production use; no new
+  failure mode introduced.
+
+## Testing
+
+In `src/paperless_mail/tests/test_mail.py`:
+
+- Extend the `BogusMailBox` test double with a `uids()` method that mirrors
+  its existing `fetch()` criteria matching but returns only UIDs.
+- New test: an attachment-less message under an attachments-only mark-read
+  rule is recorded as `PROCESSED_WO_CONSUMPTION` on the first run. On a
+  second run, with the same message still unseen server-side, assert no
+  body fetch happens (only `M.uids()` is called) -- i.e. steady state costs
+  one search and zero body downloads.
+- New test: a folder with more new UIDs than `MAIL_FETCH_BATCH_SIZE` results
+  in multiple batched `fetch` calls, and all messages are still processed
+  (no messages dropped at a batch boundary).
+- Update existing `test_handle_empty_message` to assert a
+  `PROCESSED_WO_CONSUMPTION` row now exists for the no-attachment case.
+- Regression: existing tests for `EML_ONLY`/`EVERYTHING` scopes and the
+  multi-rule-same-folder dedup (`consumed_messages` set) should pass
+  unchanged, since neither the per-message `already_processed` check nor the
+  `consumed_messages` set logic is touched by this design.
+
+## Open questions / risks
+
+- `imap_tools`'s `AND(uid=batch)` criteria builder needs to be confirmed to
+  produce a valid `UID FETCH <list>` command at the batch sizes used here;
+  covered by the batching test above.
+- If a mail server doesn't support `UID SEARCH` the same way it supports the
+  existing `fetch`'s implicit search, `M.uids()` could behave differently
+  from today's `M.fetch()` on some edge-case server. Existing project test
+  coverage (via `BogusMailBox`) won't catch server-specific quirks; this is
+  the same class of risk as any IMAP-behavior assumption already baked into
+  this module.

src/paperless_mail/mail.py

@@ -73,6 +73,8 @@ APPLE_MAIL_TAG_COLORS = {
     "grey": ["$MailFlagBit1", "$MailFlagBit2"],
 }
 
+MAIL_FETCH_BATCH_SIZE = 500
+
 
 class MailError(Exception):
     pass
@@ -681,17 +683,55 @@ class MailAccountHandler(LoggingMixin):
         )
 
         try:
-            messages = M.fetch(
+            all_uids = set(
-                criteria=criterias,
+                M.uids(criteria=criterias, charset=rule.account.character_set),
-                mark_seen=False,
-                charset=rule.account.character_set,
-                bulk=True,
             )
         except Exception as err:
             raise MailError(
-                f"Rule {rule}: Error while fetching folder {rule.folder}",
+                f"Rule {rule}: Error while searching folder {rule.folder}",
             ) from err
 
+        processed_uids_qs = ProcessedMail.objects.filter(
+            rule=rule,
+            folder=rule.folder,
+            uid__in=all_uids,
+        )
+        if self._current_uid_validity is not None:
+            processed_uids_qs = processed_uids_qs.filter(
+                Q(uid_validity=self._current_uid_validity)
+                | Q(uid_validity__isnull=True),
+            )
+        processed_uids = set(processed_uids_qs.values_list("uid", flat=True))
+
+        new_uids = all_uids - processed_uids
+
+        if not new_uids:
+            self.log.debug(
+                f"Rule {rule}: No new mail matching criteria {criterias}",
+            )
+            return 0
+
+        sorted_new_uids = sorted(new_uids, key=int)
+        message_batches = []
+        # ikvk/imap_tools#268 requests a direct UID-list fetch that would let us drop this manual batching loop
+        for batch_start in range(0, len(sorted_new_uids), MAIL_FETCH_BATCH_SIZE):
+            batch = sorted_new_uids[batch_start : batch_start + MAIL_FETCH_BATCH_SIZE]
+            try:
+                message_batches.append(
+                    M.fetch(
+                        criteria=AND(uid=batch),
+                        mark_seen=False,
+                        charset=rule.account.character_set,
+                        bulk=True,
+                    ),
+                )
+            except Exception as err:
+                raise MailError(
+                    f"Rule {rule}: Error while fetching folder {rule.folder}",
+                ) from err
+
+        messages = itertools.chain(*message_batches)
+
         mails_processed = 0
         total_processed_files = 0
         rule_seen_messages: set[tuple[str, str | None]] = set()
@@ -757,6 +797,7 @@ class MailAccountHandler(LoggingMixin):
             not message.attachments
             and rule.consumption_scope == MailRule.ConsumptionScope.ATTACHMENTS_ONLY
         ):
+            self._record_processed_without_consumption(message, rule)
             return processed_elements
 
         self.log.debug(
@@ -792,6 +833,25 @@ class MailAccountHandler(LoggingMixin):
 
         return processed_elements
 
+    def _record_processed_without_consumption(
+        self,
+        message: MailMessage,
+        rule: MailRule,
+    ) -> None:
+        ProcessedMail.objects.get_or_create(
+            rule=rule,
+            uid=message.uid,
+            folder=rule.folder,
+            uid_validity=self._current_uid_validity,
+            defaults={
+                "subject": message.subject,
+                "received": make_aware(message.date)
+                if is_naive(message.date)
+                else message.date,
+                "status": "PROCESSED_WO_CONSUMPTION",
+            },
+        )
+
     def filename_inclusion_matches(
         self,
         filter_attachment_filename_include: str | None,
@@ -958,23 +1018,7 @@ class MailAccountHandler(LoggingMixin):
             )
         else:
             # No files to consume, just mark as processed if it wasn't by .eml processing
-            if not ProcessedMail.objects.filter(
+            self._record_processed_without_consumption(message, rule)
-                rule=rule,
-                uid=message.uid,
-                folder=rule.folder,
-                uid_validity=self._current_uid_validity,
-            ).exists():
-                ProcessedMail.objects.create(
-                    rule=rule,
-                    folder=rule.folder,
-                    uid=message.uid,
-                    uid_validity=self._current_uid_validity,
-                    subject=message.subject,
-                    received=make_aware(message.date)
-                    if is_naive(message.date)
-                    else message.date,
-                    status="PROCESSED_WO_CONSUMPTION",
-                )
 
         return processed_attachments
 

src/paperless_mail/tests/test_mail.py

@@ -135,6 +135,12 @@ class BogusMailBox(AbstractContextManager):
             raise MailboxLoginError("BAD", "OK")
 
     def fetch(self, criteria, mark_seen, charset="", *, bulk=True):
+        return self._filter_messages(criteria)
+
+    def uids(self, criteria, charset="") -> list[str]:
+        return [m.uid for m in self._filter_messages(criteria)]
+
+    def _filter_messages(self, criteria):
         msg = self.messages
 
         criteria = str(criteria).strip("()").split(" ")
@@ -168,6 +174,10 @@ class BogusMailBox(AbstractContextManager):
         if "(X-GM-LABELS" in criteria:  # ['NOT', '(X-GM-LABELS', '"processed"']
             msg = filter(lambda m: "processed" not in m.flags, msg)
 
+        if "UID" in criteria:
+            uid_list = criteria[criteria.index("UID") + 1].split(",")
+            msg = filter(lambda m: m.uid in uid_list, msg)
+
         return list(msg)
 
     def delete(self, uid_list) -> None:
@@ -425,6 +435,54 @@ class TestMail(
 
         super().setUp()
 
+    @mock.patch("paperless_mail.mail.MAIL_FETCH_BATCH_SIZE", 5)
+    def test_handle_mail_account_batches_body_fetch_for_large_backlog(self) -> None:
+        """
+        GIVEN:
+            - More new/unprocessed mail than MAIL_FETCH_BATCH_SIZE
+        WHEN:
+            - The mail account is processed
+        THEN:
+            - The body fetch is issued in multiple batches
+            - Every message is still processed (none dropped at a batch boundary)
+        """
+        account = MailAccount.objects.create(
+            name="test",
+            imap_server="",
+            username="admin",
+            password="secret",
+        )
+        rule = MailRule.objects.create(
+            name="testrule",
+            account=account,
+            action=MailRule.MailAction.MARK_READ,
+            consumption_scope=MailRule.ConsumptionScope.ATTACHMENTS_ONLY,
+        )
+
+        message_count = 12  # more than the patched batch size of 5
+        self.mailMocker.bogus_mailbox.messages = [
+            self.mailMocker.messageBuilder.create_message(
+                subject=f"No attachment {i}",
+                attachments=[],
+            )
+            for i in range(message_count)
+        ]
+        self.mailMocker.bogus_mailbox.updateClient()
+
+        with mock.patch.object(
+            self.mailMocker.bogus_mailbox,
+            "fetch",
+            wraps=self.mailMocker.bogus_mailbox.fetch,
+        ) as fetch_spy:
+            self.mail_account_handler.handle_mail_account(account)
+
+        # ceil(12 / 5) == 3 batches
+        self.assertEqual(fetch_spy.call_count, 3)
+        self.assertEqual(
+            ProcessedMail.objects.filter(rule=rule).count(),
+            message_count,
+        )
+
     def test_get_correspondent(self) -> None:
         message = namedtuple("MailMessage", [])
         message.from_ = "someone@somewhere.com"
@@ -537,17 +595,59 @@ class TestMail(
             ],
         )
 
-    def test_handle_empty_message(self) -> None:
+    def test_bogus_mailbox_uids_and_uid_criteria(self) -> None:
-        message = namedtuple("MailMessage", [])
+        mailbox = self.mailMocker.bogus_mailbox
+        all_messages = list(mailbox.messages)
 
-        message.attachments = []
+        # uids() returns the UIDs of unseen messages, no bodies needed to call it
-        rule = MailRule()
+        unseen_uids = mailbox.uids("(UNSEEN)")
+        self.assertEqual(
+            set(unseen_uids),
+            {m.uid for m in all_messages if not m.seen},
+        )
+
+        # fetch() with an explicit UID criteria returns only the matching messages
+        target_uid = all_messages[0].uid
+        from imap_tools import AND
+
+        fetched = mailbox.fetch(AND(uid=[target_uid]), mark_seen=False)
+        self.assertEqual([m.uid for m in fetched], [target_uid])
+
+    def test_handle_empty_message(self) -> None:
+        message = self.mailMocker.messageBuilder.create_message(
+            subject="No attachments here",
+            attachments=[],
+        )
+
+        account = MailAccount.objects.create()
+        rule = MailRule.objects.create(
+            account=account,
+            consumption_scope=MailRule.ConsumptionScope.ATTACHMENTS_ONLY,
+        )
 
         result = self.mail_account_handler._handle_message(message, rule)
 
         self.mailMocker._queue_consumption_tasks_mock.assert_not_called()
         self.assertEqual(result, 0)
 
+        processed = ProcessedMail.objects.get(
+            rule=rule,
+            uid=message.uid,
+            folder=rule.folder,
+        )
+        self.assertEqual(processed.status, "PROCESSED_WO_CONSUMPTION")
+
+        # Calling it again must not create a second row
+        self.mail_account_handler._handle_message(message, rule)
+        self.assertEqual(
+            ProcessedMail.objects.filter(
+                rule=rule,
+                uid=message.uid,
+                folder=rule.folder,
+            ).count(),
+            1,
+        )
+
     def test_handle_unknown_mime_type(self) -> None:
         message = self.mailMocker.messageBuilder.create_message(
             attachments=[
@@ -912,6 +1012,62 @@ class TestMail(
         ]
         self.assertEqual(queued_rule.id, first_rule.id)
 
+    def test_handle_mail_account_skips_body_fetch_for_already_processed_mail(
+        self,
+    ) -> None:
+        """
+        GIVEN:
+            - An attachment-less mail under an attachments-only mark-read rule,
+              already recorded as PROCESSED_WO_CONSUMPTION
+        WHEN:
+            - The mail account is processed again and the mail still matches the
+              search criteria (it was never marked read, since no mail action is
+              applied for the no-consumption case)
+        THEN:
+            - No IMAP body fetch happens for that mail; only the cheap UID search runs.
+        """
+        account = MailAccount.objects.create(
+            name="test",
+            imap_server="",
+            username="admin",
+            password="secret",
+        )
+        rule = MailRule.objects.create(
+            name="testrule",
+            account=account,
+            action=MailRule.MailAction.MARK_READ,
+            consumption_scope=MailRule.ConsumptionScope.ATTACHMENTS_ONLY,
+        )
+
+        message = self.mailMocker.messageBuilder.create_message(
+            subject="No attachment",
+            attachments=[],
+        )
+        self.mailMocker.bogus_mailbox.messages = [message]
+        self.mailMocker.bogus_mailbox.updateClient()
+
+        # First run: records ProcessedMail without consuming anything.
+        self.mail_account_handler.handle_mail_account(account)
+        self.assertTrue(
+            ProcessedMail.objects.filter(
+                rule=rule,
+                uid=message.uid,
+                folder=rule.folder,
+            ).exists(),
+        )
+        self.mailMocker._queue_consumption_tasks_mock.assert_not_called()
+
+        # Second run: message still matches UNSEEN (mark-read action never ran),
+        # but its body must not be downloaded again.
+        with mock.patch.object(
+            self.mailMocker.bogus_mailbox,
+            "fetch",
+            wraps=self.mailMocker.bogus_mailbox.fetch,
+        ) as fetch_spy:
+            self.mail_account_handler.handle_mail_account(account)
+
+        fetch_spy.assert_not_called()
+
     def test_handle_mail_account_skip_duplicate_uids_from_fetch(self) -> None:
         """
         GIVEN: