typed UnsupportedSourceError replaces pipeline string-marker matching

haiku_rag_slim/haiku/rag/client/documents.py

@@ -5,6 +5,7 @@ from urllib.parse import unquote, urlparse
 
 import logfire
 
+from haiku.rag.client.exceptions import UnsupportedSourceError
 from haiku.rag.client.processing import (
     ensure_chunks_embedded,
     get_extension_from_content_type_or_url,
@@ -226,7 +227,7 @@ async def _ingest_fetch_result(
         result.uri, result.content_type
     )
     if file_extension not in converter.supported_extensions:
-        raise ValueError(
+        raise UnsupportedSourceError(
             f"Unsupported content type/extension: {result.content_type}/{file_extension}"
         )
 
@@ -338,7 +339,7 @@ async def create_document_from_source(
         )
         if local_path.is_dir():
             if uri is not None:
-                raise ValueError(
+                raise UnsupportedSourceError(
                     "uri override is not supported for directory sources; each file "
                     "produces its own document with its own auto-derived URI."
                 )
@@ -359,13 +360,15 @@ async def create_document_from_source(
             return documents
 
         if not local_path.exists():
-            raise ValueError(f"File does not exist: {local_path}")
+            raise UnsupportedSourceError(f"File does not exist: {local_path}")
 
         # Match the old _create_document_from_file behaviour: fail fast on
         # unsupported extension before reading any bytes.
         converter = get_converter(client._config)
         if local_path.suffix.lower() not in converter.supported_extensions:
-            raise ValueError(f"Unsupported file extension: {local_path.suffix}")
+            raise UnsupportedSourceError(
+                f"Unsupported file extension: {local_path.suffix}"
+            )
 
     # Single resource — resolve the right Source adapter for this URI.
     fetcher = resolve_fetcher(source_str, storage_options=storage_options)

haiku_rag_slim/haiku/rag/client/exceptions.py

@@ -0,0 +1,7 @@
+class UnsupportedSourceError(ValueError):
+    """A source URI or file cannot be ingested and never will be on a retry —
+    unsupported extension, unsupported content type, missing file, malformed
+    S3 URI, etc. ``ValueError`` for backward compatibility with callers that
+    used to catch a plain ValueError; the ingester pipeline catches the
+    specific type to classify as ``PermanentError`` without string matching.
+    """

haiku_rag_slim/haiku/rag/client/processing.py

@@ -6,6 +6,7 @@ from urllib.parse import urlparse
 
 import httpx
 
+from haiku.rag.client.exceptions import UnsupportedSourceError
 from haiku.rag.config import AppConfig
 from haiku.rag.converters import get_converter
 from haiku.rag.store.models.chunk import Chunk
@@ -100,9 +101,9 @@ async def convert(
     # Path object - convert file directly
     if isinstance(source, Path):
         if not source.exists():
-            raise ValueError(f"File does not exist: {source}")
+            raise UnsupportedSourceError(f"File does not exist: {source}")
         if source.suffix.lower() not in converter.supported_extensions:
-            raise ValueError(f"Unsupported file extension: {source.suffix}")
+            raise UnsupportedSourceError(f"Unsupported file extension: {source.suffix}")
         effective_uri = source_uri or source.absolute().as_uri()
         doc = await _convert_file(source, effective_uri)
         _warn_if_descriptions_missing(config, doc, str(source))
@@ -123,7 +124,7 @@ async def convert(
             )
 
             if file_extension not in converter.supported_extensions:
-                raise ValueError(
+                raise UnsupportedSourceError(
                     f"Unsupported content type/extension: {content_type}/{file_extension}"
                 )
 
@@ -146,9 +147,11 @@ async def convert(
         # file:// URI
         file_path = Path(parsed.path)
         if not file_path.exists():
-            raise ValueError(f"File does not exist: {file_path}")
+            raise UnsupportedSourceError(f"File does not exist: {file_path}")
         if file_path.suffix.lower() not in converter.supported_extensions:
-            raise ValueError(f"Unsupported file extension: {file_path.suffix}")
+            raise UnsupportedSourceError(
+                f"Unsupported file extension: {file_path.suffix}"
+            )
         effective_uri = source_uri or file_path.absolute().as_uri()
         doc = await _convert_file(file_path, effective_uri)
         _warn_if_descriptions_missing(config, doc, str(file_path))

haiku_rag_slim/haiku/rag/ingester/sources/registry.py

@@ -2,6 +2,7 @@ from collections.abc import Iterable
 from pathlib import Path
 from urllib.parse import urlparse
 
+from haiku.rag.client.exceptions import UnsupportedSourceError
 from haiku.rag.ingester.sources.base import Source
 from haiku.rag.ingester.sources.fs import FSSource
 from haiku.rag.ingester.sources.http import HTTPSource
@@ -36,7 +37,7 @@ def resolve_fetcher(
     if scheme == "s3":
         bucket = urlparse(uri).netloc
         if not bucket:
-            raise ValueError(f"Invalid S3 URI: {uri}")
+            raise UnsupportedSourceError(f"Invalid S3 URI: {uri}")
         return S3Source(uri=f"s3://{bucket}/", storage_options=storage_options)
 
-    raise ValueError(f"No source adapter for URI scheme {scheme!r}: {uri}")
+    raise UnsupportedSourceError(f"No source adapter for URI scheme {scheme!r}: {uri}")

haiku_rag_slim/haiku/rag/ingester/sources/s3.py

@@ -4,6 +4,7 @@ from collections.abc import AsyncIterator
 from datetime import UTC, datetime
 from urllib.parse import urlparse
 
+from haiku.rag.client.exceptions import UnsupportedSourceError
 from haiku.rag.ingester.sources.base import (
     FetchResult,
     RevisionSnapshot,
@@ -19,7 +20,7 @@ from haiku.rag.ingester.sources.filter import (
 def _parse_s3_uri(uri: str) -> tuple[str, str]:
     parsed = urlparse(uri)
     if parsed.scheme != "s3" or not parsed.netloc:
-        raise ValueError(f"Invalid S3 URI: {uri}")
+        raise UnsupportedSourceError(f"Invalid S3 URI: {uri}")
     return parsed.netloc, parsed.path.lstrip("/")
 
 
@@ -27,7 +28,7 @@ def _parse_s3_object_uri(uri: str) -> tuple[str, str]:
     """Like _parse_s3_uri but rejects bucket-only URIs (no key)."""
     bucket, key = _parse_s3_uri(uri)
     if not key:
-        raise ValueError(f"Invalid S3 URI: {uri}")
+        raise UnsupportedSourceError(f"Invalid S3 URI: {uri}")
     return bucket, key
 
 

haiku_rag_slim/haiku/rag/ingester/workers/pipeline.py

@@ -6,22 +6,13 @@ import httpx
 import logfire
 from pydantic import BaseModel
 
+from haiku.rag.client.exceptions import UnsupportedSourceError
 from haiku.rag.ingester.exceptions import PermanentError, TransientError
 from haiku.rag.ingester.queue.models import Job, JobOp
 
 if TYPE_CHECKING:
     from haiku.rag.client import HaikuRAG
 
-# ValueError messages from create_document_from_source that mean "this job
-# will never succeed". Anything else from ValueError defaults to transient.
-_PERMANENT_VALUE_MARKERS = (
-    "Unsupported file extension",
-    "Unsupported content type",
-    "Invalid S3 URI",
-    "File does not exist",
-    "uri override is not supported",
-)
-
 
 class JobResult(BaseModel):
     """What the worker needs after a successful job: enough metadata to
@@ -39,11 +30,17 @@ def _classify(exc: BaseException) -> Exception:
     if isinstance(exc, PermanentError | TransientError):
         return exc
 
+    # UnsupportedSourceError is the typed signal from client/* that the
+    # source will never ingest successfully on a retry (bad URI scheme,
+    # missing file, unsupported extension, etc.).
+    if isinstance(exc, UnsupportedSourceError):
+        return PermanentError(str(exc))
+
     if isinstance(exc, ValueError):
-        message = str(exc)
-        if any(marker in message for marker in _PERMANENT_VALUE_MARKERS):
-            return PermanentError(message)
-        return TransientError(message)
+        # Some downstream libraries (e.g. docling) raise plain ValueError
+        # for "couldn't parse this file"; default to transient so the queue
+        # retries up to max_attempts in case the issue is intermittent.
+        return TransientError(str(exc))
 
     if isinstance(exc, httpx.HTTPStatusError):
         status = exc.response.status_code

tests/ingester/test_pipeline.py

@@ -104,8 +104,10 @@ async def test_delete_is_noop_when_document_missing():
 
 @pytest.mark.asyncio
 async def test_unsupported_extension_classified_permanent():
+    from haiku.rag.client.exceptions import UnsupportedSourceError
+
     client = _mock_client()
-    client.create_document_from_source.side_effect = ValueError(
+    client.create_document_from_source.side_effect = UnsupportedSourceError(
         "Unsupported file extension: .xyz"
     )
 
@@ -115,8 +117,10 @@ async def test_unsupported_extension_classified_permanent():
 
 @pytest.mark.asyncio
 async def test_unsupported_content_type_classified_permanent():
+    from haiku.rag.client.exceptions import UnsupportedSourceError
+
     client = _mock_client()
-    client.create_document_from_source.side_effect = ValueError(
+    client.create_document_from_source.side_effect = UnsupportedSourceError(
         "Unsupported content type/extension: application/octet-stream/.bin"
     )
     with pytest.raises(PermanentError):
@@ -125,8 +129,10 @@ async def test_unsupported_content_type_classified_permanent():
 
 @pytest.mark.asyncio
 async def test_invalid_s3_uri_classified_permanent():
+    from haiku.rag.client.exceptions import UnsupportedSourceError
+
     client = _mock_client()
-    client.create_document_from_source.side_effect = ValueError(
+    client.create_document_from_source.side_effect = UnsupportedSourceError(
         "Invalid S3 URI: s3:///bad"
     )
     with pytest.raises(PermanentError):
@@ -135,8 +141,10 @@ async def test_invalid_s3_uri_classified_permanent():
 
 @pytest.mark.asyncio
 async def test_missing_file_classified_permanent():
+    from haiku.rag.client.exceptions import UnsupportedSourceError
+
     client = _mock_client()
-    client.create_document_from_source.side_effect = ValueError(
+    client.create_document_from_source.side_effect = UnsupportedSourceError(
         "File does not exist: /nope"
     )
     with pytest.raises(PermanentError):