Removes the plan

imap_tools 1.14.0 adds a uid_list fetch arg that fetches by UID
directly without issuing a SEARCH command, and does its own
bulk-fetch batching. Use it in place of the manual per-batch
AND(uid=...) fetch loop.

pyproject.toml

@@ -47,7 +47,7 @@ dependencies = [
   "gotenberg-client~=0.14.0",
   "httpx-oauth~=0.16",
   "ijson>=3.2",
-  "imap-tools~=1.13.0",
+  "imap-tools~=1.14.0",
   "jinja2~=3.1.5",
   "langdetect~=1.0.9",
   "llama-index-core>=0.14.22",

src/documents/search/_query.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import logging
+from datetime import UTC
 from typing import TYPE_CHECKING
 from typing import Final
 
@@ -8,6 +9,12 @@ import regex
 import tantivy
 from django.conf import settings
 
+from documents.search._dates import (
+    _date_only_range,  # noqa: F401 — re-exported for test imports
+)
+from documents.search._dates import (
+    _datetime_range,  # noqa: F401 — re-exported for test imports
+)
 from documents.search._tokenizer import simple_search_tokens
 from documents.search._translate import SearchQueryError
 from documents.search._translate import translate_query
@@ -69,6 +76,42 @@ def _build_cjk_query(
         return None
 
 
+def rewrite_natural_date_keywords(query: str, tz: tzinfo) -> str:
+    """
+    Rewrite natural date syntax to ISO 8601 format for Tantivy compatibility.
+
+    Delegates to ``translate_query`` which handles all date forms, comma
+    expansion, field aliasing, relative ranges, and operator normalization.
+
+    Args:
+        query: Raw user query string
+        tz: Timezone for converting local date boundaries to UTC
+
+    Returns:
+        Query with date syntax rewritten to ISO 8601 ranges
+
+    Note:
+        Bare keywords without field prefixes pass through unchanged.
+    """
+    return translate_query(query, tz)
+
+
+def normalize_query(query: str) -> str:
+    """
+    Normalize query syntax for better search behavior.
+
+    Delegates to ``translate_query`` which handles comma expansion, whitespace
+    collapsing, operator normalization, and field aliasing.
+
+    Args:
+        query: Query string after date rewriting
+
+    Returns:
+        Normalized query string ready for Tantivy parsing
+    """
+    return translate_query(query, UTC)
+
+
 def build_permission_filter(
     schema: tantivy.Schema,
     user: AbstractBaseUser,

src/documents/tests/search/test_query.py

@@ -11,15 +11,16 @@ import pytest
 import tantivy
 import time_machine
 
-from documents.search._dates import _date_only_range
+from documents.search._query import _date_only_range
-from documents.search._dates import _datetime_range
+from documents.search._query import _datetime_range
 from documents.search._query import build_permission_filter
+from documents.search._query import normalize_query
 from documents.search._query import parse_simple_text_highlight_query
 from documents.search._query import parse_user_query
+from documents.search._query import rewrite_natural_date_keywords
 from documents.search._schema import build_schema
 from documents.search._tokenizer import register_tokenizers
 from documents.search._translate import InvalidDateQuery
-from documents.search._translate import translate_query
 
 if TYPE_CHECKING:
     from django.contrib.auth.base_user import AbstractBaseUser
@@ -58,7 +59,7 @@ class TestCreatedDateField:
     )
     @time_machine.travel(datetime(2026, 3, 28, 15, 30, tzinfo=UTC), tick=False)
     def test_today(self, tz: tzinfo, expected_lo: str, expected_hi: str) -> None:
-        lo, hi = _range(translate_query("created:today", tz), "created")
+        lo, hi = _range(rewrite_natural_date_keywords("created:today", tz), "created")
         assert lo == expected_lo
         assert hi == expected_hi
 
@@ -66,7 +67,7 @@ class TestCreatedDateField:
     def test_today_auckland_ahead_of_utc(self) -> None:
         # UTC 03:00 -> Auckland (UTC+13) = 16:00 same date; local date = 2026-03-28
         lo, _ = _range(
-            translate_query("created:today", AUCKLAND),
+            rewrite_natural_date_keywords("created:today", AUCKLAND),
             "created",
         )
         assert lo == "2026-03-28T00:00:00Z"
@@ -128,7 +129,7 @@ class TestCreatedDateField:
     ) -> None:
         # 2026-03-28 is Saturday; Mon-Sun week calculation built into expectations
         query = f"{field}:{keyword}"
-        lo, hi = _range(translate_query(query, UTC), field)
+        lo, hi = _range(rewrite_natural_date_keywords(query, UTC), field)
         assert lo == expected_lo
         assert hi == expected_hi
 
@@ -136,7 +137,7 @@ class TestCreatedDateField:
     def test_this_month_december_wraps_to_next_year(self) -> None:
         # December: next month must roll over to January 1 of next year
         lo, hi = _range(
-            translate_query("created:this month", UTC),
+            rewrite_natural_date_keywords("created:this month", UTC),
             "created",
         )
         assert lo == "2026-12-01T00:00:00Z"
@@ -146,7 +147,7 @@ class TestCreatedDateField:
     def test_last_month_january_wraps_to_previous_year(self) -> None:
         # January: last month must roll back to December 1 of previous year
         lo, hi = _range(
-            translate_query("created:previous month", UTC),
+            rewrite_natural_date_keywords("created:previous month", UTC),
             "created",
         )
         assert lo == "2025-12-01T00:00:00Z"
@@ -155,7 +156,7 @@ class TestCreatedDateField:
     @time_machine.travel(datetime(2026, 7, 15, 12, 0, tzinfo=UTC), tick=False)
     def test_previous_quarter(self) -> None:
         lo, hi = _range(
-            translate_query('created:"previous quarter"', UTC),
+            rewrite_natural_date_keywords('created:"previous quarter"', UTC),
             "created",
         )
         assert lo == "2026-04-01T00:00:00Z"
@@ -175,7 +176,7 @@ class TestDateTimeFields:
     @time_machine.travel(datetime(2026, 3, 28, 15, 30, tzinfo=UTC), tick=False)
     def test_added_today_eastern(self) -> None:
         # EDT = UTC-4; local midnight 2026-03-28 00:00 EDT = 2026-03-28 04:00 UTC
-        lo, hi = _range(translate_query("added:today", EASTERN), "added")
+        lo, hi = _range(rewrite_natural_date_keywords("added:today", EASTERN), "added")
         assert lo == "2026-03-28T04:00:00Z"
         assert hi == "2026-03-29T04:00:00Z"
 
@@ -183,14 +184,14 @@ class TestDateTimeFields:
     def test_added_today_auckland_midnight_crossing(self) -> None:
         # UTC 02:00 on 2026-03-29 -> Auckland (UTC+13) = 2026-03-29 15:00 local
         # Auckland midnight = UTC 2026-03-28 11:00
-        lo, hi = _range(translate_query("added:today", AUCKLAND), "added")
+        lo, hi = _range(rewrite_natural_date_keywords("added:today", AUCKLAND), "added")
         assert lo == "2026-03-28T11:00:00Z"
         assert hi == "2026-03-29T11:00:00Z"
 
     @time_machine.travel(datetime(2026, 3, 28, 15, 0, tzinfo=UTC), tick=False)
     def test_modified_today_utc(self) -> None:
         lo, hi = _range(
-            translate_query("modified:today", UTC),
+            rewrite_natural_date_keywords("modified:today", UTC),
             "modified",
         )
         assert lo == "2026-03-28T00:00:00Z"
@@ -245,14 +246,14 @@ class TestDateTimeFields:
         expected_hi: str,
     ) -> None:
         # 2026-03-28 is Saturday; weekday()==5 so Monday=2026-03-23
-        lo, hi = _range(translate_query(f"added:{keyword}", UTC), "added")
+        lo, hi = _range(rewrite_natural_date_keywords(f"added:{keyword}", UTC), "added")
         assert lo == expected_lo
         assert hi == expected_hi
 
     @time_machine.travel(datetime(2026, 12, 15, 12, 0, tzinfo=UTC), tick=False)
     def test_this_month_december_wraps_to_next_year(self) -> None:
         # December: next month wraps to January of next year
-        lo, hi = _range(translate_query("added:this month", UTC), "added")
+        lo, hi = _range(rewrite_natural_date_keywords("added:this month", UTC), "added")
         assert lo == "2026-12-01T00:00:00Z"
         assert hi == "2027-01-01T00:00:00Z"
 
@@ -260,7 +261,7 @@ class TestDateTimeFields:
     def test_last_month_january_wraps_to_previous_year(self) -> None:
         # January: last month wraps back to December of previous year
         lo, hi = _range(
-            translate_query("added:previous month", UTC),
+            rewrite_natural_date_keywords("added:previous month", UTC),
             "added",
         )
         assert lo == "2025-12-01T00:00:00Z"
@@ -296,7 +297,7 @@ class TestDateTimeFields:
         expected_lo: str,
         expected_hi: str,
     ) -> None:
-        lo, hi = _range(translate_query(query, UTC), "added")
+        lo, hi = _range(rewrite_natural_date_keywords(query, UTC), "added")
         assert lo == expected_lo
         assert hi == expected_hi
 
@@ -310,20 +311,20 @@ class TestWhooshQueryRewriting:
 
     @time_machine.travel(datetime(2026, 3, 28, 15, 0, tzinfo=UTC), tick=False)
     def test_compact_date_shim_rewrites_to_iso(self) -> None:
-        result = translate_query("created:20240115120000", UTC)
+        result = rewrite_natural_date_keywords("created:20240115120000", UTC)
         assert "2024-01-15" in result
         assert "20240115120000" not in result
 
     @time_machine.travel(datetime(2026, 3, 28, 15, 0, tzinfo=UTC), tick=False)
     def test_relative_range_shim_removes_now(self) -> None:
-        result = translate_query("added:[now-7d TO now]", UTC)
+        result = rewrite_natural_date_keywords("added:[now-7d TO now]", UTC)
         assert "now" not in result
         assert "2026-03-" in result
 
     @time_machine.travel(datetime(2026, 3, 28, 12, 0, tzinfo=UTC), tick=False)
     def test_bracket_minus_7_days(self) -> None:
         lo, hi = _range(
-            translate_query("added:[-7 days to now]", UTC),
+            rewrite_natural_date_keywords("added:[-7 days to now]", UTC),
             "added",
         )
         assert lo == "2026-03-21T12:00:00Z"
@@ -332,7 +333,7 @@ class TestWhooshQueryRewriting:
     @time_machine.travel(datetime(2026, 3, 28, 12, 0, tzinfo=UTC), tick=False)
     def test_bracket_minus_1_week(self) -> None:
         lo, hi = _range(
-            translate_query("added:[-1 week to now]", UTC),
+            rewrite_natural_date_keywords("added:[-1 week to now]", UTC),
             "added",
         )
         assert lo == "2026-03-21T12:00:00Z"
@@ -342,7 +343,7 @@ class TestWhooshQueryRewriting:
     def test_bracket_minus_1_month_uses_relativedelta(self) -> None:
         # relativedelta(months=1) from 2026-03-28 = 2026-02-28 (not 29)
         lo, hi = _range(
-            translate_query("created:[-1 month to now]", UTC),
+            rewrite_natural_date_keywords("created:[-1 month to now]", UTC),
             "created",
         )
         assert lo == "2026-02-28T12:00:00Z"
@@ -351,7 +352,7 @@ class TestWhooshQueryRewriting:
     @time_machine.travel(datetime(2026, 3, 28, 12, 0, tzinfo=UTC), tick=False)
     def test_bracket_minus_1_year(self) -> None:
         lo, hi = _range(
-            translate_query("modified:[-1 year to now]", UTC),
+            rewrite_natural_date_keywords("modified:[-1 year to now]", UTC),
             "modified",
         )
         assert lo == "2025-03-28T12:00:00Z"
@@ -360,7 +361,7 @@ class TestWhooshQueryRewriting:
     @time_machine.travel(datetime(2026, 3, 28, 12, 0, tzinfo=UTC), tick=False)
     def test_bracket_plural_unit_hours(self) -> None:
         lo, hi = _range(
-            translate_query("added:[-3 hours to now]", UTC),
+            rewrite_natural_date_keywords("added:[-3 hours to now]", UTC),
             "added",
         )
         assert lo == "2026-03-28T09:00:00Z"
@@ -368,7 +369,7 @@ class TestWhooshQueryRewriting:
 
     @time_machine.travel(datetime(2026, 3, 28, 12, 0, tzinfo=UTC), tick=False)
     def test_bracket_case_insensitive(self) -> None:
-        result = translate_query("added:[-1 WEEK TO NOW]", UTC)
+        result = rewrite_natural_date_keywords("added:[-1 WEEK TO NOW]", UTC)
         assert "now" not in result.lower()
         lo, hi = _range(result, "added")
         assert lo == "2026-03-21T12:00:00Z"
@@ -378,7 +379,7 @@ class TestWhooshQueryRewriting:
     def test_relative_range_swaps_bounds_when_lo_exceeds_hi(self) -> None:
         # [now+1h TO now-1h] has lo > hi before substitution; they must be swapped
         lo, hi = _range(
-            translate_query("added:[now+1h TO now-1h]", UTC),
+            rewrite_natural_date_keywords("added:[now+1h TO now-1h]", UTC),
             "added",
         )
         assert lo == "2026-03-28T11:00:00Z"
@@ -386,14 +387,14 @@ class TestWhooshQueryRewriting:
 
     def test_8digit_created_date_field_always_uses_utc_midnight(self) -> None:
         # created is a DateField: boundaries are always UTC midnight, no TZ offset
-        result = translate_query("created:20231201", EASTERN)
+        result = rewrite_natural_date_keywords("created:20231201", EASTERN)
         lo, hi = _range(result, "created")
         assert lo == "2023-12-01T00:00:00Z"
         assert hi == "2023-12-02T00:00:00Z"
 
     def test_8digit_added_datetime_field_converts_local_midnight_to_utc(self) -> None:
         # added is DateTimeField: midnight Dec 1 Eastern (EST = UTC-5) = 05:00 UTC
-        result = translate_query("added:20231201", EASTERN)
+        result = rewrite_natural_date_keywords("added:20231201", EASTERN)
         lo, hi = _range(result, "added")
         assert lo == "2023-12-01T05:00:00Z"
         assert hi == "2023-12-02T05:00:00Z"
@@ -401,7 +402,7 @@ class TestWhooshQueryRewriting:
     def test_8digit_modified_datetime_field_converts_local_midnight_to_utc(
         self,
     ) -> None:
-        result = translate_query("modified:20231201", EASTERN)
+        result = rewrite_natural_date_keywords("modified:20231201", EASTERN)
         lo, hi = _range(result, "modified")
         assert lo == "2023-12-01T05:00:00Z"
         assert hi == "2023-12-02T05:00:00Z"
@@ -411,7 +412,7 @@ class TestWhooshQueryRewriting:
         # (e.g. month=13) so the API can surface a 400 telling the user the date
         # is malformed instead of silently returning zero results.
         with pytest.raises(InvalidDateQuery) as exc_info:
-            translate_query("added:20231340", UTC)
+            rewrite_natural_date_keywords("added:20231340", UTC)
         assert exc_info.value.field == "added"
         assert exc_info.value.value == "20231340"
 
@@ -578,7 +579,7 @@ class TestYearRangeRewriting:
         expected_lo: str,
         expected_hi: str,
     ) -> None:
-        result = translate_query(query, UTC)
+        result = rewrite_natural_date_keywords(query, UTC)
         lo, hi = _range(result, field)
         assert lo == expected_lo
         assert hi == expected_hi
@@ -586,14 +587,14 @@ class TestYearRangeRewriting:
     def test_reversed_year_range_is_swapped(self) -> None:
         # A reversed range must not yield lo > hi, which Tantivy treats as an
         # empty range (silently zero results). The bounds are swapped instead.
-        result = translate_query("created:[2025 TO 2020]", UTC)
+        result = rewrite_natural_date_keywords("created:[2025 TO 2020]", UTC)
         lo, hi = _range(result, "created")
         assert lo == "2020-01-01T00:00:00Z"
         assert hi == "2026-01-01T00:00:00Z"
 
     def test_year_range_in_complex_boolean_query(self) -> None:
         query = "tag:steuer AND (title:2020 OR (NOT title:2019 AND NOT title:2018 AND created:[2020 TO 2020]))"
-        result = translate_query(query, UTC)
+        result = rewrite_natural_date_keywords(query, UTC)
         lo, hi = _range(result, "created")
         assert lo == "2020-01-01T00:00:00Z"
         assert hi == "2021-01-01T00:00:00Z"
@@ -603,7 +604,7 @@ class TestYearRangeRewriting:
 
     def test_already_iso_date_range_passes_through_unchanged(self) -> None:
         original = "created:[2020-01-01T00:00:00Z TO 2021-01-01T00:00:00Z]"
-        assert translate_query(original, UTC) == original
+        assert rewrite_natural_date_keywords(original, UTC) == original
 
     def test_8digit_in_brackets_not_matched_as_year_range(self) -> None:
         # [YYYYMMDD TO YYYYMMDD]: the translation layer converts 8-digit bounds to
@@ -612,7 +613,7 @@ class TestYearRangeRewriting:
         # This is the correct and accepted behavior: old compact form becomes a
         # proper Tantivy-parseable ISO range.
         original = "created:[20200101 TO 20201231]"
-        result = translate_query(original, UTC)
+        result = rewrite_natural_date_keywords(original, UTC)
         lo, hi = _range(result, "created")
         assert lo == "2020-01-01T00:00:00Z"
         assert hi == "2021-01-01T00:00:00Z"
@@ -635,7 +636,7 @@ class TestNonDateFieldsNotRewritten:
         ],
     )
     def test_8digit_on_integer_field_passes_through_unchanged(self, query: str) -> None:
-        assert translate_query(query, EASTERN) == query
+        assert rewrite_natural_date_keywords(query, EASTERN) == query
 
     @pytest.mark.parametrize(
         "query",
@@ -649,12 +650,12 @@ class TestNonDateFieldsNotRewritten:
         self,
         query: str,
     ) -> None:
-        assert translate_query(query, UTC) == query
+        assert rewrite_natural_date_keywords(query, UTC) == query
 
     def test_unknown_field_keyword_passes_through_unchanged(self) -> None:
         # foobar is not a date field: 'foobar:today' must not become a date range,
         # which Tantivy would otherwise reject as an unknown/typed field.
-        assert translate_query("foobar:today", UTC) == "foobar:today"
+        assert rewrite_natural_date_keywords("foobar:today", UTC) == "foobar:today"
 
 
 class TestPassthrough:
@@ -662,39 +663,37 @@ class TestPassthrough:
 
     def test_bare_keyword_no_field_prefix_unchanged(self) -> None:
         # Bare 'today' with no field: prefix passes through unchanged
-        result = translate_query("bank statement today", UTC)
+        result = rewrite_natural_date_keywords("bank statement today", UTC)
         assert "today" in result
 
     def test_unrelated_query_unchanged(self) -> None:
-        assert translate_query("title:invoice", UTC) == "title:invoice"
+        assert rewrite_natural_date_keywords("title:invoice", UTC) == "title:invoice"
 
 
 class TestNormalizeQuery:
-    """translate_query expands comma-separated values and collapses whitespace."""
+    """normalize_query expands comma-separated values and collapses whitespace."""
 
     def test_normalize_expands_comma_separated_tags(self) -> None:
-        assert translate_query("tag:foo,bar", UTC) == "tag:foo AND tag:bar"
+        assert normalize_query("tag:foo,bar") == "tag:foo AND tag:bar"
 
     def test_normalize_comma_between_range_expressions(self) -> None:
         # Comma-separated field range expressions (Whoosh v2 syntax) must be
         # converted to AND so Tantivy does not receive an invalid comma.
         q = "created:[2026-01-01T00:00:00Z TO 2026-06-01T00:00:00Z],added:[2026-05-01T00:00:00Z TO 2026-06-01T00:00:00Z]"
-        assert translate_query(q, UTC) == (
+        assert normalize_query(q) == (
             "created:[2026-01-01T00:00:00Z TO 2026-06-01T00:00:00Z]"
             " AND "
             "added:[2026-05-01T00:00:00Z TO 2026-06-01T00:00:00Z]"
         )
 
     def test_normalize_expands_three_values(self) -> None:
-        assert (
+        assert normalize_query("tag:foo,bar,baz") == "tag:foo AND tag:bar AND tag:baz"
-            translate_query("tag:foo,bar,baz", UTC) == "tag:foo AND tag:bar AND tag:baz"
-        )
 
     def test_normalize_collapses_whitespace(self) -> None:
-        assert translate_query("bank   statement", UTC) == "bank statement"
+        assert normalize_query("bank   statement") == "bank statement"
 
     def test_normalize_no_commas_unchanged(self) -> None:
-        assert translate_query("bank statement", UTC) == "bank statement"
+        assert normalize_query("bank statement") == "bank statement"
 
     @pytest.mark.parametrize(
         ("raw", "expected"),
@@ -737,7 +736,7 @@ class TestNormalizeQuery:
         ],
     )
     def test_normalize_strips_dangling_operators(self, raw: str, expected: str) -> None:
-        assert translate_query(raw, UTC) == expected
+        assert normalize_query(raw) == expected
 
     @pytest.mark.parametrize(
         "query",
@@ -749,7 +748,7 @@ class TestNormalizeQuery:
         ],
     )
     def test_normalize_preserves_valid_operators(self, query: str) -> None:
-        assert translate_query(query, UTC) == query
+        assert normalize_query(query) == query
 
 
 class TestParseSimpleTextHighlightQuery:

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
@@ -680,12 +682,41 @@ class MailAccountHandler(LoggingMixin):
             f"Rule {rule}: Searching folder with criteria {criterias}",
         )
 
+        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
+
+        sorted_new_uids = sorted(new_uids, key=int)
         try:
             messages = M.fetch(
-                criteria=criterias,
+                uid_list=sorted_new_uids,
                 mark_seen=False,
-                charset=rule.account.character_set,
+                bulk=MAIL_FETCH_BATCH_SIZE,
-                bulk=True,
             )
         except Exception as err:
             raise MailError(
@@ -757,6 +788,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 +824,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 +1009,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

@@ -134,7 +134,23 @@ class BogusMailBox(AbstractContextManager):
         if username != self.USERNAME or access_token != self.ACCESS_TOKEN:
             raise MailboxLoginError("BAD", "OK")
 
-    def fetch(self, criteria, mark_seen, charset="", *, bulk=True):
+    def fetch(
+        self,
+        criteria="ALL",
+        charset="",
+        *,
+        mark_seen=True,
+        bulk=True,
+        uid_list=None,
+    ):
+        if uid_list is not None:
+            return [m for m in self.messages if m.uid in uid_list]
+        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 +184,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:
@@ -406,7 +426,7 @@ def assert_eventually_equals(
     deadline = time.time() + timeout
     while time.time() < deadline:
         if getter_fn() == expected_value:
-            return None
+            return
         time.sleep(interval)
     actual = getter_fn()
     raise AssertionError(f"Expected {expected_value}, but got {actual}")
@@ -425,6 +445,58 @@ 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 once, with all UIDs and the configured batch size
+              handed to imap_tools so it can bulk-fetch in batches server-side
+            - 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)
+
+        # A single fetch() call hands the full UID list and batch size to imap_tools,
+        # which does its own bulk-fetching in batches of MAIL_FETCH_BATCH_SIZE.
+        fetch_spy.assert_called_once()
+        self.assertEqual(fetch_spy.call_args.kwargs["bulk"], 5)
+        self.assertEqual(len(fetch_spy.call_args.kwargs["uid_list"]), message_count)
+        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 +609,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 +1026,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:
@@ -1517,7 +1687,7 @@ class TestMail(
             if message.from_ == "amazon@amazon.de":
                 raise ValueError("Does not compute.")
             else:
-                return None
+                return
 
         m.side_effect = get_correspondent_fake