Minor improvements from a Claude review

ChatStreamingView built an IN filter from every permitted document id
for the "chat over all documents" case, which exceeds the vector
store's SQLite bound-parameter safety limit on installs with more
than ~32700 documents, silently returning no context. For a user who
can see every document (an active superuser), that filter never
narrows anything, so skip it and let the retriever search the whole
index instead.

src/documents/management/commands/document_consumer.py

@@ -314,7 +314,7 @@ def _consume_file(
     consumption_dir: Path,
     *,
     subdirs_as_tags: bool,
-) -> bool:
+) -> None:
     """
     Queue a file for consumption.
 
@@ -322,20 +322,15 @@ def _consume_file(
         filepath: Path to the file to consume.
         consumption_dir: Base consumption directory.
         subdirs_as_tags: Whether to create tags from subdirectory names.
-
-    Returns:
-        True if the file was successfully handed to Celery, False otherwise.
-        Callers must not record the file as queued on failure, or the rescan
-        will never retry it.
     """
     # Verify file still exists and is accessible
     try:
         if not filepath.is_file():
             logger.debug(f"Not consuming {filepath}: not a file or doesn't exist")
-            return False
+            return
     except OSError as e:
         logger.warning(f"Not consuming {filepath}: {e}")
-        return False
+        return
 
     # Get tags from path if configured
     tag_ids: list[int] | None = None
@@ -360,9 +355,6 @@ def _consume_file(
         )
     except Exception:
         logger.exception(f"Error while queuing document {filepath}")
-        return False
-
-    return True
 
 
 class Command(BaseCommand):
@@ -500,12 +492,12 @@ class Command(BaseCommand):
             if not consumer_filter(Change.added, str(filepath)):
                 continue
 
-            if _consume_file(
+            _consume_file(
                 filepath=filepath,
                 consumption_dir=directory,
                 subdirs_as_tags=subdirs_as_tags,
-            ):
-                queued.add(filepath.resolve())
+            )
+            queued.add(filepath.resolve())
 
         return queued
 
@@ -659,16 +651,14 @@ class Command(BaseCommand):
 
                     # Check for stable files
                     for stable_path in tracker.get_stable_files():
-                        # Only remember files that were actually queued, so the
-                        # rescan does not re-queue them while the consume task
-                        # has yet to remove them from disk, but does retry a
-                        # failed publish instead of stranding it
-                        if _consume_file(
+                        _consume_file(
                             filepath=stable_path,
                             consumption_dir=directory,
                             subdirs_as_tags=subdirs_as_tags,
-                        ):
-                            queued.add(stable_path)
+                        )
+                        # Remember it so the rescan does not re-queue it while
+                        # the consume task has yet to remove it from disk
+                        queued.add(stable_path)
 
                     # Exit watch loop to reconfigure timeout
                     break

src/documents/tests/test_api_chat.py

@@ -1,11 +1,18 @@
 from __future__ import annotations
 
+from typing import TYPE_CHECKING
 from unittest import mock
 
+import pytest
+from django.contrib.auth.models import Permission
 from django.contrib.auth.models import User
 from rest_framework import status
+from rest_framework.test import APIClient
 from rest_framework.test import APITestCase
 
+if TYPE_CHECKING:
+    from pytest_mock import MockerFixture
+
 
 class TestChatStreamingViewInputValidation(APITestCase):
     def setUp(self) -> None:
@@ -42,3 +49,70 @@ class TestChatStreamingViewInputValidation(APITestCase):
                 format="json",
             )
         assert resp.status_code == status.HTTP_400_BAD_REQUEST
+
+
+@pytest.mark.django_db
+class TestChatStreamingViewUnrestrictedFlag:
+    """The document id filter may only be skipped (``unrestricted=True``) for
+    a caller who can see every document, i.e. an active superuser.
+    """
+
+    @pytest.fixture
+    def mocked_stream_chat(self, mocker: MockerFixture) -> mock.MagicMock:
+        """AI enabled, with stream_chat_with_documents patched so the view
+        never touches the real vector store; returns the patched callable so
+        tests can inspect how it was called.
+        """
+        mocker.patch("documents.views.AIConfig").return_value.ai_enabled = True
+        return mocker.patch(
+            "documents.views.stream_chat_with_documents",
+            return_value=iter(()),
+        )
+
+    @pytest.fixture
+    def viewer_client(self, user_client: APIClient, regular_user: User) -> APIClient:
+        """The conftest regular-user client, additionally granted
+        view_document -- able to see every document without being a
+        superuser.
+        """
+        regular_user.user_permissions.add(
+            *Permission.objects.filter(codename="view_document"),
+        )
+        return user_client
+
+    @pytest.mark.parametrize(
+        ("client_fixture", "expected_unrestricted"),
+        [
+            pytest.param("admin_client", True, id="superuser_is_unrestricted"),
+            pytest.param("viewer_client", False, id="regular_user_is_restricted"),
+        ],
+    )
+    def test_unrestricted_only_for_superuser(
+        self,
+        request: pytest.FixtureRequest,
+        mocked_stream_chat: mock.MagicMock,
+        client_fixture: str,
+        *,
+        expected_unrestricted: bool,
+    ) -> None:
+        """
+        GIVEN:
+            - A superuser, or a regular user holding view_document
+        WHEN:
+            - They post a chat question with no document_id
+        THEN:
+            - stream_chat_with_documents is called with unrestricted=True for
+              the superuser and unrestricted=False for the regular user, even
+              though that user can view every document
+        """
+        client: APIClient = request.getfixturevalue(client_fixture)
+
+        client.post(
+            "/api/documents/chat/",
+            data={"q": "What's in these documents?"},
+            format="json",
+        )
+
+        assert (
+            mocked_stream_chat.call_args.kwargs["unrestricted"] is expected_unrestricted
+        )

src/documents/tests/test_management_consumer.py

@@ -445,13 +445,12 @@ class TestConsumeFile:
         target = consumption_dir / "document.pdf"
         shutil.copy(sample_pdf, target)
 
-        result = _consume_file(
+        _consume_file(
             filepath=target,
             consumption_dir=consumption_dir,
             subdirs_as_tags=False,
         )
 
-        assert result is True
         mock_consume_file_delay.apply_async.assert_called_once()
         call_args = mock_consume_file_delay.apply_async.call_args
         consumable_doc = call_args.kwargs["kwargs"]["input_doc"]
@@ -465,12 +464,11 @@ class TestConsumeFile:
         mock_consume_file_delay: MagicMock,
     ) -> None:
         """Test _consume_file handles nonexistent files gracefully."""
-        result = _consume_file(
+        _consume_file(
             filepath=consumption_dir / "nonexistent.pdf",
             consumption_dir=consumption_dir,
             subdirs_as_tags=False,
         )
-        assert result is False
         mock_consume_file_delay.apply_async.assert_not_called()
 
     def test_consume_directory(
@@ -482,12 +480,11 @@ class TestConsumeFile:
         subdir = consumption_dir / "subdir"
         subdir.mkdir()
 
-        result = _consume_file(
+        _consume_file(
             filepath=subdir,
             consumption_dir=consumption_dir,
             subdirs_as_tags=False,
         )
-        assert result is False
         mock_consume_file_delay.apply_async.assert_not_called()
 
     def test_consume_with_permission_error(
@@ -502,33 +499,13 @@ class TestConsumeFile:
         shutil.copy(sample_pdf, target)
 
         mocker.patch.object(Path, "is_file", side_effect=PermissionError("denied"))
-        result = _consume_file(
+        _consume_file(
             filepath=target,
             consumption_dir=consumption_dir,
             subdirs_as_tags=False,
         )
-        assert result is False
         mock_consume_file_delay.apply_async.assert_not_called()
 
-    def test_consume_with_apply_async_failure(
-        self,
-        consumption_dir: Path,
-        sample_pdf: Path,
-        mock_consume_file_delay: MagicMock,
-    ) -> None:
-        """Test _consume_file reports failure when apply_async raises."""
-        target = consumption_dir / "document.pdf"
-        shutil.copy(sample_pdf, target)
-
-        mock_consume_file_delay.apply_async.side_effect = Exception("broker down")
-
-        result = _consume_file(
-            filepath=target,
-            consumption_dir=consumption_dir,
-            subdirs_as_tags=False,
-        )
-        assert result is False
-
     def test_consume_with_tags_error(
         self,
         consumption_dir: Path,
@@ -545,12 +522,11 @@ class TestConsumeFile:
             side_effect=DatabaseError("Something happened"),
         )
 
-        result = _consume_file(
+        _consume_file(
             filepath=target,
             consumption_dir=consumption_dir,
             subdirs_as_tags=True,
         )
-        assert result is True
         mock_consume_file_delay.apply_async.assert_called_once()
         call_args = mock_consume_file_delay.apply_async.call_args
         overrides = call_args.kwargs["kwargs"]["overrides"]
@@ -1273,52 +1249,6 @@ class TestProcessExistingFilesQueued:
         assert target.resolve() in queued
 
 
-@pytest.mark.management
-@pytest.mark.django_db
-class TestCommandRetryAfterQueueFailure:
-    """
-    Regression test for GH #13923.
-
-    A file whose ``apply_async`` publish fails (e.g. broker briefly down)
-    must not be marked as queued, so the periodic rescan retries it once
-    the broker recovers, instead of stranding it until the consumer
-    process is restarted.
-    """
-
-    def test_watch_loop_retries_failed_publish_on_rescan(
-        self,
-        consumption_dir: Path,
-        sample_pdf: Path,
-        mock_consume_file_delay: MagicMock,
-        start_consumer: Callable[..., ConsumerThread],
-    ) -> None:
-        """A publish failure from the watch loop is retried by the rescan."""
-        apply_async = mock_consume_file_delay.apply_async
-
-        def fail_first_call(*args: object, **kwargs: object) -> None:
-            if apply_async.call_count == 1:
-                raise Exception("broker down")
-
-        apply_async.side_effect = fail_first_call
-
-        thread = start_consumer(stability_delay=0.1, rescan_interval=0.3)
-
-        target = consumption_dir / "document.pdf"
-        shutil.copy(sample_pdf, target)
-
-        deadline = monotonic() + 5.0
-        while apply_async.call_count < 2 and monotonic() < deadline:
-            sleep(0.1)
-
-        if thread.exception:
-            raise thread.exception
-
-        assert apply_async.call_count >= 2, (
-            "Expected the failed publish to be retried by the rescan, "
-            f"but apply_async was only called {apply_async.call_count} time(s)"
-        )
-
-
 @pytest.mark.management
 @pytest.mark.django_db
 class TestCommandRescanRecovery:

src/documents/views.py

@@ -180,6 +180,7 @@ from documents.permissions import has_system_status_permission
 from documents.permissions import permitted_document_ids
 from documents.permissions import permitted_object_ids
 from documents.permissions import set_permissions_for_object
+from documents.permissions import user_is_unrestricted
 from documents.plugins.date_parsing import get_date_parser
 from documents.schema import generate_object_with_permissions_schema
 from documents.search import SearchHit
@@ -2329,10 +2330,12 @@ class ChatStreamingView(GenericAPIView[Any]):
                 return HttpResponseForbidden("Insufficient permissions")
 
             documents = Document.objects.filter(pk=document.pk)
+            unrestricted = False
         else:
             documents = Document.objects.filter(
                 id__in=permitted_document_ids(request.user),
             )
+            unrestricted = user_is_unrestricted(request.user)
 
         output_language = get_llm_output_language(
             ai_config=ai_config,
@@ -2343,6 +2346,7 @@ class ChatStreamingView(GenericAPIView[Any]):
             stream_chat_with_documents(
                 query_str=question,
                 documents=documents,
+                unrestricted=unrestricted,
                 output_language=output_language,
             ),
             content_type="text/event-stream",

src/paperless/settings/__init__.py

@@ -705,12 +705,6 @@ CELERY_BROKER_TRANSPORT_OPTIONS = {
 CELERY_TASK_TRACK_STARTED = True
 CELERY_TASK_TIME_LIMIT: Final[int] = get_int_from_env("PAPERLESS_WORKER_TIMEOUT", 1800)
 
-# https://docs.celeryq.dev/en/stable/userguide/configuration.html#std-setting-task_allow_error_cb_on_chord_header
-# Without this, a failing chord header never triggers the errback, so a mail
-# whose attachments all fail is never recorded and is re-fetched forever.
-# The errback runs once per failed header task, so it must be idempotent.
-CELERY_TASK_ALLOW_ERROR_CB_ON_CHORD_HEADER = True
-
 CELERY_CACHE_BACKEND = "default"
 
 # https://docs.celeryq.dev/en/stable/userguide/configuration.html#task-serializer

src/paperless_ai/chat.py

@@ -95,12 +95,15 @@ def _format_chat_metadata_trailer(references: list[dict[str, int | str]]) -> str
 def stream_chat_with_documents(
     query_str: str,
     documents: QuerySet[Document],
+    *,
+    unrestricted: bool = False,
     output_language: str | None = None,
 ):
     try:
         yield from _stream_chat_with_documents(
             query_str,
             documents,
+            unrestricted=unrestricted,
             output_language=output_language,
         )
     except Exception as e:
@@ -111,6 +114,8 @@ def stream_chat_with_documents(
 def _stream_chat_with_documents(
     query_str: str,
     documents: QuerySet[Document],
+    *,
+    unrestricted: bool = False,
     output_language: str | None = None,
 ):
     if not documents.exists():
@@ -123,9 +128,15 @@ def _stream_chat_with_documents(
     from llama_index.core.retrievers import VectorIndexRetriever
 
     config = AIConfig()
-    filters = _document_id_filters(
-        str(pk) for pk in documents.values_list("pk", flat=True)
-    )
+    if unrestricted:
+        # The caller can see every document, so an id filter would never narrow
+        # the search, only risk exceeding the vector store's bound parameter
+        # limit (_MAX_IN_VALUES in vector_store.py) on large installs.
+        filters = None
+    else:
+        filters = _document_id_filters(
+            str(pk) for pk in documents.values_list("pk", flat=True)
+        )
 
     # Hold the shared read lock for the whole operation: the query engine
     # retrieves from the vector store again during synthesis, so the connection

src/paperless_ai/tests/test_chat.py

@@ -1,4 +1,8 @@
+from __future__ import annotations
+
 import json
+from typing import TYPE_CHECKING
+from typing import Any
 from unittest.mock import MagicMock
 from unittest.mock import patch
 
@@ -18,6 +22,11 @@ from paperless_ai.chat import _build_chat_prompt
 from paperless_ai.chat import _build_refine_prompt
 from paperless_ai.chat import stream_chat_with_documents
 
+if TYPE_CHECKING:
+    from pathlib import Path
+
+    import pytest_mock
+
 
 @pytest.fixture(autouse=True)
 def patch_embed_model():
@@ -312,6 +321,30 @@ def test_stream_chat_unexpected_failure_returns_generic_error(caplog) -> None:
 
 @pytest.mark.django_db
 class TestStreamChatRetrieval:
+    @pytest.fixture
+    def captured_filters(self, mocker: pytest_mock.MockerFixture) -> list[Any]:
+        """Stub out the AI client and the retriever, capturing the ``filters``
+        kwarg of every VectorIndexRetriever construction.
+
+        VectorIndexRetriever is imported inside _stream_chat_with_documents,
+        so it is patched at the llama_index source for the lazy import to
+        pick it up.
+        """
+        captured: list[Any] = []
+        retriever = mocker.MagicMock()
+        retriever.retrieve.return_value = []
+
+        def capture_retriever(*args, **kwargs) -> pytest_mock.MockType:
+            captured.append(kwargs.get("filters"))
+            return retriever
+
+        mocker.patch("paperless_ai.chat.AIClient")
+        mocker.patch(
+            "llama_index.core.retrievers.VectorIndexRetriever",
+            side_effect=capture_retriever,
+        )
+        return captured
+
     def test_no_nodes_yields_no_content_message(
         self,
         temp_llm_index_dir,
@@ -329,9 +362,9 @@ class TestStreamChatRetrieval:
 
     def test_chat_filter_contains_only_requested_document_ids(
         self,
-        temp_llm_index_dir,
-        mock_embed_model,
-        mocker,
+        temp_llm_index_dir: Path,
+        mock_embed_model: pytest_mock.MockType,
+        captured_filters: list[Any],
     ) -> None:
         """The MetadataFilter passed to the retriever must be scoped to the
         requested documents only — content from other indexed documents must
@@ -342,22 +375,6 @@ class TestStreamChatRetrieval:
         indexing.llm_index_add_or_update_document(included)
         indexing.llm_index_add_or_update_document(excluded)
 
-        # VectorIndexRetriever is imported inside _stream_chat_with_documents;
-        # patch it at the llama_index source so the lazy import picks it up.
-        captured_filters = []
-        mock_retriever = mocker.MagicMock()
-        mock_retriever.retrieve.return_value = []
-
-        def capture_retriever(*args, **kwargs):
-            captured_filters.append(kwargs.get("filters"))
-            return mock_retriever
-
-        mocker.patch("paperless_ai.chat.AIClient")
-        mocker.patch(
-            "llama_index.core.retrievers.VectorIndexRetriever",
-            side_effect=capture_retriever,
-        )
-
         list(
             chat.stream_chat_with_documents(
                 "question?",
@@ -372,6 +389,36 @@ class TestStreamChatRetrieval:
         assert str(included.pk) in filter_values
         assert str(excluded.pk) not in filter_values
 
+    def test_unrestricted_chat_skips_document_id_filter(
+        self,
+        temp_llm_index_dir: Path,
+        mock_embed_model: pytest_mock.MockType,
+        captured_filters: list[Any],
+    ) -> None:
+        """
+        GIVEN:
+            - A document indexed in the vector store
+        WHEN:
+            - stream_chat_with_documents is called with unrestricted=True
+        THEN:
+            - The retriever receives no document id filter (filters=None), so
+              the whole index is searched instead of an IN-list that risks the
+              vector store's safety limit on large installs
+        """
+        document = DocumentFactory.create(content="indexed document content")
+        indexing.llm_index_add_or_update_document(document)
+
+        list(
+            chat.stream_chat_with_documents(
+                "question?",
+                Document.objects.filter(pk=document.pk),
+                unrestricted=True,
+            ),
+        )
+
+        assert captured_filters, "VectorIndexRetriever was never constructed"
+        assert captured_filters[0] is None
+
     @pytest.mark.django_db
     def test_get_document_references_only_queries_referenced_documents(
         self,

src/paperless_mail/mail.py

@@ -334,24 +334,18 @@ def error_callback(
     """
     A shared task that is called whenever something goes wrong during
     consumption of a file. See queue_consumption_tasks.
-
-    With CELERY_TASK_ALLOW_ERROR_CB_ON_CHORD_HEADER enabled this runs once per
-    failed header task, not once per chord, so it must be idempotent.
     """
     rule = MailRule.objects.get(pk=rule_id)
-    received = make_aware(message_date) if is_naive(message_date) else message_date
 
-    ProcessedMail.objects.get_or_create(
+    ProcessedMail.objects.create(
         rule=rule,
         folder=rule.folder,
         uid=message_uid,
         uid_validity=uid_validity,
-        defaults={
-            "subject": message_subject,
-            "received": received,
-            "status": "FAILED",
-            "error": traceback.format_exc(),
-        },
+        subject=message_subject,
+        received=make_aware(message_date) if is_naive(message_date) else message_date,
+        status="FAILED",
+        error=traceback.format_exc(),
     )
 
 

src/paperless_mail/tests/test_mail.py

@@ -36,7 +36,6 @@ from paperless_mail.mail import MailAccountHandler
 from paperless_mail.mail import MailError
 from paperless_mail.mail import TagMailAction
 from paperless_mail.mail import apply_mail_action
-from paperless_mail.mail import error_callback
 from paperless_mail.mail import get_mailbox
 from paperless_mail.models import MailAccount
 from paperless_mail.models import MailRule
@@ -2046,44 +2045,6 @@ class TestPostConsumeAction(TestCase):
         self.assertIn("Test Exception", processed_mail.error)
 
 
-@pytest.mark.django_db
-class TestErrorCallback:
-    def test_error_callback_is_idempotent_for_same_mail(self) -> None:
-        """
-        GIVEN:
-            - A mail rule and a mail that failed to be consumed
-        WHEN:
-            - error_callback is invoked more than once for the same mail, as
-              happens when task_allow_error_cb_on_chord_header fires the
-              errback once per failed header task in a chord
-        THEN:
-            - Only one ProcessedMail row is created for that mail
-        """
-        rule = MailRuleFactory()
-        message_uid = "12345"
-
-        for _ in range(2):
-            error_callback(
-                None,
-                Exception("Test Exception"),
-                None,
-                rule_id=rule.pk,
-                message_uid=message_uid,
-                message_subject="Test Subject",
-                message_date=timezone.make_aware(
-                    timezone.datetime(2023, 1, 1, 12, 0, 0),
-                ),
-            )
-
-        processed_mails = ProcessedMail.objects.filter(
-            rule=rule,
-            uid=message_uid,
-            folder=rule.folder,
-        )
-        assert processed_mails.count() == 1
-        assert processed_mails.get().status == "FAILED"
-
-
 class TestManagementCommand(TestCase):
     @mock.patch(
         "paperless_mail.management.commands.mail_fetcher.tasks.process_mail_accounts",