Merge pull request '🐾 fix(tasks): unsupported file types fail ingest terminally, no retries' (#8) from fix/unsupported-file-type-terminal into main

Reviewed-on: #8

docs/deploy.md

@@ -85,26 +85,21 @@ an explicit `when: mnemosyne_first_deploy` flag.
 
 ```bash
 # Apply Django ORM migrations (PostgreSQL schema)
-docker compose -f /srv/mnemosyne/docker-compose.yaml run --rm app migrate
+docker compose run --rm app migrate
 
 # Create Neo4j vector + full-text indexes and load library-type defaults
-docker compose -f /srv/mnemosyne/docker-compose.yaml \
+docker compose run --rm app setup
-    run --rm app setup
 
 # Seed the MCPSigningKey used to sign long-lived Pallas team JWTs.
 # --retire-other deactivates any previously-active key.  The hex
 # emitted to stdout is persisted in Mnemosyne's database and is
 # not re-injected from the vault — no operator action required
 # beyond running this command once per fresh deployment.
-docker compose -f /srv/mnemosyne/docker-compose.yaml \
+docker compose run --rm app python manage.py seed_signing_key --kid daedalus-1 --retire-other
-    run --rm app \
-    python manage.py seed_signing_key --kid daedalus-1 --retire-other
 
 # Create Django groups for SSO role mapping (View Only / Staff / SME / Admin).
 # Safe to re-run — idempotent.
-docker compose -f /srv/mnemosyne/docker-compose.yaml \
+docker compose run --rm app python manage.py create_sso_groups
-    run --rm app \
-    python manage.py create_sso_groups
 ```
 
 The `seed_signing_key` command prints the generated secret once to stdout — it

mnemosyne/library/services/parsers.py

@@ -22,6 +22,17 @@ from .text_utils import remove_excessive_whitespace, sanitize_text
 
 logger = logging.getLogger(__name__)
 
+
+class UnsupportedFileTypeError(ValueError):
+    """Raised when a file's type cannot be parsed.
+
+    Deterministic and input-driven — re-parsing identical bytes can never
+    succeed — so ingest must treat it as a terminal failure and never retry.
+    Subclasses ``ValueError`` so existing ``except ValueError`` callers still
+    catch it.
+    """
+
+
 # File extensions supported by PyMuPDF
 PYMUPDF_EXTENSIONS = {
     "pdf", "epub", "xps", "mobi", "fb2", "cbz", "svg",
@@ -88,7 +99,7 @@ class DocumentParser:
         :param file_path: Path to the document file.
         :param file_type: File extension (without dot), e.g. 'pdf', 'epub'.
         :returns: ParseResult with text blocks, images, and metadata.
-        :raises ValueError: If the file type is not supported.
+        :raises UnsupportedFileTypeError: If the file type is not supported.
         """
         file_type = file_type.lower().lstrip(".")
 
@@ -117,7 +128,7 @@ class DocumentParser:
         if file_type in ("html", "htm"):
             return self._parse_with_pymupdf(file_path, file_type)
 
-        raise ValueError(
+        raise UnsupportedFileTypeError(
             f"Unsupported file type '{file_type}'. "
             f"Supported: {sorted(PYMUPDF_EXTENSIONS | PLAINTEXT_EXTENSIONS | IMAGE_EXTENSIONS)}"
         )

mnemosyne/library/tasks.py

@@ -347,6 +347,7 @@ def ingest_from_daedalus(self, job_id: str):
     from datetime import datetime, timezone
 
     from library.models import IngestJob, Item, Library
+    from library.services.parsers import UnsupportedFileTypeError
     from library.services.source_s3 import (
         copy_into_mnemosyne,
         fetch_from_source,
@@ -465,6 +466,24 @@ def ingest_from_daedalus(self, job_id: str):
             **result,
         }
 
+    except UnsupportedFileTypeError as exc:
+        # Deterministic, input-driven — re-parsing identical bytes can never
+        # succeed. Terminal client-data failure, never retried, and logged at
+        # WARNING (not ERROR) because an unparseable input is not a server fault.
+        logger.warning(
+            "Task ingest_from_daedalus rejected job_id=%s: %s", job_id, exc,
+        )
+        job.status = "failed"
+        job.error = str(exc)
+        job.completed_at = datetime.now(timezone.utc)
+        job.save(update_fields=["status", "error", "completed_at"])
+        return {
+            "success": False,
+            "job_id": job_id,
+            "error": str(exc),
+            "reason": "unsupported_file_type",
+        }
+
     except Exception as exc:
         logger.error(
             "Task ingest_from_daedalus failed job_id=%s: %s",

mnemosyne/library/tests/test_tasks.py

@@ -67,6 +67,114 @@ class ReembedItemTaskTests(TestCase):
         mock_pipeline.reprocess_item.assert_called_once()
 
 
+@override_settings(CELERY_TASK_ALWAYS_EAGER=True)
+class IngestFromDaedalusFailureClassificationTests(TestCase):
+    """A deterministic parse failure is terminal; a transient error retries.
+
+    Neo4j and S3 are mocked at the task's boundaries so the test exercises the
+    exception-classification branch (parsers.UnsupportedFileTypeError vs. any
+    other Exception) without a live graph or object store.
+    """
+
+    def setUp(self):
+        from library.tasks import ingest_from_daedalus
+
+        # Calling .run() bypasses Celery's request setup, but the task
+        # persists self.request.id into the NOT NULL celery_task_id column
+        # and reads self.request.retries — push a real request context.
+        ingest_from_daedalus.push_request(id="test-task-id", retries=0)
+        self.addCleanup(ingest_from_daedalus.pop_request)
+
+    def _make_job(self):
+        from library.models import IngestJob
+
+        return IngestJob.objects.create(
+            id="job_test_unsupported",
+            library_uid="lib-uid-123",
+            source="daedalus",
+            s3_key="incoming/bad.drawio",
+            file_type="vnd.jgraph.mxfile",
+            title="bad.drawio",
+            content_hash="abc123",
+        )
+
+    def _patched_boundaries(self, pipeline_side_effect):
+        """Patch every boundary the task hits before the pipeline runs.
+
+        Returns a context-manager list; the pipeline's ``process_item`` is set
+        to raise ``pipeline_side_effect``.
+        """
+        from library.services.parsers import UnsupportedFileTypeError  # noqa: F401
+
+        patchers = [
+            patch("library.tasks.db"),
+            patch("library.models.Library"),
+            patch("library.models.Item"),
+            patch("library.services.source_s3.fetch_from_source", return_value=b"data"),
+            patch("library.services.source_s3.copy_into_mnemosyne"),
+            patch("library.tasks._resolve_or_create_default_collection"),
+            patch("library.services.pipeline.EmbeddingPipeline"),
+        ]
+        mocks = [p.start() for p in patchers]
+        self.addCleanup(lambda: [p.stop() for p in patchers])
+
+        # db.cypher_query returns (rows, meta); no prior item to supersede.
+        mocks[0].cypher_query.return_value = ([], None)
+        # Library.nodes.get returns a stand-in library node.
+        mocks[1].nodes.get.return_value = MagicMock(uid="lib-uid-123")
+        # Item() instances carry a uid used for the S3 key.
+        item_instance = MagicMock(uid="item-uid-abc")
+        mocks[2].return_value = item_instance
+        # The pipeline raises the classification-relevant error.
+        pipeline_instance = MagicMock()
+        pipeline_instance.process_item.side_effect = pipeline_side_effect
+        mocks[6].return_value = pipeline_instance
+        return pipeline_instance
+
+    def test_unsupported_file_type_is_terminal_and_not_retried(self):
+        from library.models import IngestJob
+        from library.services.parsers import UnsupportedFileTypeError
+        from library.tasks import ingest_from_daedalus
+
+        job = self._make_job()
+        self._patched_boundaries(
+            UnsupportedFileTypeError("Unsupported file type 'vnd.jgraph.mxfile'.")
+        )
+
+        with patch.object(ingest_from_daedalus, "retry") as mock_retry:
+            result = ingest_from_daedalus.run(job.id)
+
+        # Never retried.
+        mock_retry.assert_not_called()
+        # Terminal failure with a machine-readable reason.
+        self.assertFalse(result["success"])
+        self.assertEqual(result["reason"], "unsupported_file_type")
+
+        job.refresh_from_db()
+        self.assertEqual(job.status, "failed")
+        self.assertEqual(job.retry_count, 0)
+        self.assertIsNotNone(job.completed_at)
+        self.assertIn("Unsupported file type", job.error)
+
+    def test_transient_error_takes_the_retry_path(self):
+        from library.tasks import ingest_from_daedalus
+
+        job = self._make_job()
+        self._patched_boundaries(ConnectionError("S3 hiccup"))
+
+        # self.retry raises Retry in real Celery; simulate that so the task
+        # doesn't fall through to the terminal branch.
+        from celery.exceptions import Retry
+
+        with patch.object(ingest_from_daedalus, "retry", side_effect=Retry()) as mock_retry:
+            with self.assertRaises(Retry):
+                ingest_from_daedalus.run(job.id)
+
+        mock_retry.assert_called_once()
+        job.refresh_from_db()
+        self.assertEqual(job.retry_count, 1)
+
+
 class ResolveUserTests(TestCase):
     """Tests for the _resolve_user helper."""