Merge pull request #858 from cbcoutinho/feat/tiered-doc-processor-b2-tier1

feat: tiered PDF processor — pypdfium2 fast path (deprecate pymupdf4llm)

nextcloud_mcp_server/config.py

@@ -136,8 +136,19 @@ _DEFAULTS: dict[str, Any] = {
     "document_pdf_graphics_limit": 1000,
     "document_parse_timeout_seconds": 120.0,
     "document_parse_mem_limit_mb": 1536,
-    # Tier-0 classifier (shadow mode: emits metrics, no routing change)
+    # Tier-0 classifier (records classification metrics on the tiered path)
     "document_classify_enabled": True,
+    # Tiered PDF pipeline: pypdfium2 is the default/only hot-path extractor;
+    # "pymupdf" is a deprecated rollback escape hatch. OCR (tier-3) is the only
+    # escalation target and is off by default (no provider wired yet).
+    "document_tier1_engine": "pypdfium2",
+    "document_ocr_enabled": False,
+    # OCR backend: "auto" picks gateway (if EMBEDDING_GATEWAY_URL) else mistral
+    # (if MISTRAL_API_KEY); "gateway"/"mistral" force one; "none" disables.
+    "document_ocr_provider": "auto",
+    # Provider-namespaced OCR model id (gateway routes on the prefix; the direct
+    # mistral backend strips it).
+    "document_ocr_model": "mistral/mistral-ocr-latest",
     # Observability
     "metrics_enabled": True,
     "metrics_port": 9090,
@@ -290,7 +301,8 @@ _dynaconf = Dynaconf(
         Validator("DOCUMENT_CHUNK_OVERLAP", gte=0),
         # Non-empty strings
         Validator("VECTOR_SYNC_PDF_TAG", len_min=1),
-        # Enum constraints
+        # Enum constraints (document_* enums are validated + normalized in
+        # __post_init__ via _enum_fields instead, for case-insensitive input).
         Validator("LOG_FORMAT", is_in=["text", "json"]),
         Validator(
             "LOG_LEVEL",
@@ -729,9 +741,22 @@ class Settings:
     # RLIMIT_AS in the parse subprocess (below the pod limit). Applied once per
     # worker for its lifetime, so changing it needs a pod restart.
     document_parse_mem_limit_mb: int = 1536
-    # Tier-0 classifier. Shadow mode for now: runs a cheap pre-pass over each PDF
-    # and emits classification metrics, but does NOT change routing yet.
+    # Tier-0 classifier. Records classification metrics (recommended_tier,
+    # text-quality) on the tiered path, derived from the tier-1 extraction.
     document_classify_enabled: bool = True
+    # PDF extraction engine for the ``fast`` tier. "pypdfium2" (default,
+    # permissive license, no find_tables) is the hot path; "pymupdf" is a
+    # deprecated rollback to pymupdf4llm (AGPL, graphics-limited) for one corpus.
+    document_tier1_engine: str = "pypdfium2"
+    # Route scanned/no-text-layer PDFs to the tier-3 OCR provider. Off by
+    # default; when off, the fast tier is terminal.
+    document_ocr_enabled: bool = False
+    # OCR backend selection: "auto" | "gateway" | "mistral" | "none".
+    document_ocr_provider: str = "auto"
+    # Provider-namespaced OCR model id (e.g. "mistral/mistral-ocr-latest"). The
+    # gateway routes on the "<provider>/" prefix; the direct mistral backend
+    # strips it.
+    document_ocr_model: str = "mistral/mistral-ocr-latest"
 
     # Observability settings
     metrics_enabled: bool = True
@@ -859,6 +884,8 @@ class Settings:
             "embedding_provider": {"autodetect", "gateway"},
             "mcp_role": {"api", "worker", "all"},
             "collection_metadata_source": {"qdrant", "api"},
+            "document_tier1_engine": {"pypdfium2", "pymupdf"},
+            "document_ocr_provider": {"auto", "gateway", "mistral", "none"},
         }
         for _field, _allowed in _enum_fields.items():
             _val = (getattr(self, _field) or "").strip().lower()
@@ -1345,6 +1372,10 @@ def get_settings() -> Settings:
         "document_parse_timeout_seconds": "DOCUMENT_PARSE_TIMEOUT_SECONDS",
         "document_parse_mem_limit_mb": "DOCUMENT_PARSE_MEM_LIMIT_MB",
         "document_classify_enabled": "DOCUMENT_CLASSIFY_ENABLED",
+        "document_tier1_engine": "DOCUMENT_TIER1_ENGINE",
+        "document_ocr_enabled": "DOCUMENT_OCR_ENABLED",
+        "document_ocr_provider": "DOCUMENT_OCR_PROVIDER",
+        "document_ocr_model": "DOCUMENT_OCR_MODEL",
         # Observability settings
         "metrics_enabled": "METRICS_ENABLED",
         "metrics_port": "METRICS_PORT",

nextcloud_mcp_server/document_processors/__init__.py

@@ -1,12 +1,20 @@
 """Document processing plugins for extracting text from various file formats."""
 
 from .base import DocumentProcessor, ProcessingResult, ProcessorError
+from .ocr import OcrProcessor
 from .pymupdf import PyMuPDFProcessor
+from .pypdfium2_fast import Pypdfium2FastProcessor
 from .registry import ProcessorRegistry, get_registry
 
-# Register processors at module initialization
+# Register processors at module initialization. The tiered PDF pipeline selects
+# by tier (not priority): Pypdfium2FastProcessor is the ``fast`` tier,
+# PyMuPDFProcessor the ``structured`` rollback, and OcrProcessor the ``ocr``
+# escalation target (reached only when document_ocr_enabled). OcrProcessor gets
+# the lowest priority so it is never the non-tiered default for PDFs.
 _registry = get_registry()
+_registry.register(Pypdfium2FastProcessor(), priority=20)
 _registry.register(PyMuPDFProcessor(), priority=10)
+_registry.register(OcrProcessor(), priority=1)
 
 __all__ = [
     "DocumentProcessor",
@@ -15,4 +23,6 @@ __all__ = [
     "ProcessorRegistry",
     "get_registry",
     "PyMuPDFProcessor",
+    "Pypdfium2FastProcessor",
+    "OcrProcessor",
 ]

nextcloud_mcp_server/document_processors/classifier.py

@@ -1,32 +1,31 @@
 """Tier-0 document classifier.
 
-A cheap (<~1s), local pre-pass over a PDF that decides which extraction tier a
-document should start in, BEFORE the expensive parse. It runs in *shadow mode*
-first: emit the signals as metrics, change no routing, and gather per-tenant
-data to tune the thresholds.
+Decides which extraction tier a PDF should escalate to, from cheap signals:
+  * text_quality -- is the text layer usable, or mashed/space-less junk? (the
+    "Student 147" lesson: a text layer can exist yet be unusable, e.g.
+    "01322234567mobile")
+  * image_coverage -- a page that is mostly a raster image is a scan/photo whose
+    content isn't fully in any text layer.
+  * no text layer -- the strongest OCR signal available from text alone.
 
-Signals (all cheap; no get_drawings, which is itself slow on the graphics-heavy
-pages we'd want to flag -- the parse-time ``graphics_limit`` already makes those
-safe, and the tier-1 quality gate catches unrecovered tables post-extraction):
+Two entry points:
+  * ``classify_from_text(text, page_boundaries)`` -- the HOT PATH. Derives the
+    text-quality/no-text-layer signal from the text the registry's tier-1 step
+    already extracted, so it adds ~no cost. No image analysis.
+  * ``classify_pdf(content)`` -- a standalone/diagnostic pass that re-opens the
+    PDF and adds image-coverage analysis. More expensive; used off the hot path.
 
-  * text_layer_chars   -- extractable text per page
-  * text_quality       -- is the text layer usable, or mashed/space-less junk?
-                          (the "Student 147" lesson: a text layer can exist yet
-                          be unusable, e.g. "01322234567mobile")
-  * image_coverage     -- fraction of the page covered by raster images
-                          (full-page image + poor text => scanned)
+Recommended tier:
+  * ``ocr``  -- scanned / no-usable-text-layer (route to tier 3, when enabled)
+  * ``fast`` -- a usable digital text layer (stay on tier 1)
 
-From these it picks a recommended starting tier:
-  * ``ocr``  -- scanned / image-only / bad-text-layer (route to tier 3)
-  * ``fast`` -- a usable digital text layer (route to tier 1)
-
-``structured`` (tier 2 / docling) is intentionally not produced here -- that tier
-is a separate service and is reached via the tier-1 quality gate, not tier-0.
+``structured`` (tier 2 / docling) is a separate service, not produced here.
 """
 
 import logging
 import re
 from dataclasses import dataclass, field
+from typing import Any
 
 logger = logging.getLogger(__name__)
 
@@ -40,6 +39,8 @@ IMAGE_COVERAGE_SCANNED = 0.80
 MIN_TEXT_QUALITY = 0.45
 # Fraction of sampled pages that must look scanned/bad for a doc->ocr verdict.
 OCR_PAGE_FRACTION = 0.5
+# A page with fewer extracted chars than this has effectively no text layer.
+MIN_PAGE_CHARS = 16
 
 _WORD_RE = re.compile(r"\S+")
 
@@ -177,3 +178,61 @@ def classify_pdf(content: bytes) -> DocClassification:
         flags=flags,
         pages=pages,
     )
+
+
+def classify_from_text(
+    full_text: str, page_boundaries: list[dict[str, Any]]
+) -> DocClassification:
+    """Classify from text already extracted by tier-1 -- no PDF re-open.
+
+    The hot-path classifier: it derives the text-quality signal from the
+    extraction the registry already ran, so it adds ~no cost (vs ``classify_pdf``,
+    which re-opens the PDF and re-extracts). It does NOT do image analysis, so it
+    cannot distinguish a scanned-with-text-layer page (that needs the image pass,
+    which only matters once OCR routing is enabled). A page with effectively no
+    text layer is the one OCR-worthy signal available from text alone.
+
+    ``page_boundaries`` are ``{page, start_offset, end_offset}`` indexing into
+    ``full_text`` (the tier-1/pdf_highlighter contract).
+    """
+    pages: list[PageSignals] = []
+    for b in page_boundaries:
+        seg = full_text[b["start_offset"] : b["end_offset"]]
+        needs_ocr = len(seg.strip()) < MIN_PAGE_CHARS
+        pages.append(
+            PageSignals(b["page"], len(seg), 0.0, _text_quality(seg), needs_ocr)
+        )
+
+    sampled = len(pages)
+    total_chars = sum(p.char_count for p in pages)
+    mean_quality = (
+        round(sum(p.text_quality for p in pages) / sampled, 3) if sampled else 0.0
+    )
+    # No pages (empty/corrupt PDF) => no OCR evidence => "fast" (the registry's
+    # page_count guard also skips escalation; defaulting to 0.0 keeps the
+    # recorded classification metric accurate rather than a misleading "ocr").
+    ocr_frac = (sum(p.needs_ocr for p in pages) / sampled) if sampled else 0.0
+
+    # Flags gated on ocr_frac >= OCR_PAGE_FRACTION (matching classify_pdf): a
+    # doc that routes "fast" must not carry a junk-layer flag just because a few
+    # isolated pages are bad -- otherwise the classification metric diverges
+    # between this hot path and the standalone classify_pdf.
+    flags: set[str] = set()
+    if sampled and ocr_frac >= OCR_PAGE_FRACTION:
+        if total_chars == 0:
+            flags.add("no_text_layer")
+        elif mean_quality < MIN_TEXT_QUALITY:
+            flags.add("bad_text_layer")
+
+    recommended = "ocr" if ocr_frac >= OCR_PAGE_FRACTION else "fast"
+
+    return DocClassification(
+        page_count=len(page_boundaries),
+        sampled_pages=sampled,
+        total_chars=total_chars,
+        mean_text_quality=mean_quality,
+        ocr_page_fraction=round(ocr_frac, 3),
+        recommended_tier=recommended,
+        flags=flags,
+        pages=pages,
+    )

nextcloud_mcp_server/document_processors/ocr.py

@@ -0,0 +1,277 @@
+"""Tier-3 OCR processor.
+
+Routes scanned / no-text-layer PDFs (the tier-0 classifier's ``ocr`` verdict) to
+an OCR backend that returns per-page markdown. Two interchangeable backends,
+selected by ``document_ocr_provider``:
+
+  * ``gateway`` -- POST the document to the Astrolabe Cloud model gateway's
+    ``POST /v1/ocr`` (the same M2M-authenticated gateway as embeddings; NO
+    provider keys in the pod). The platform default.
+  * ``mistral`` -- call the Mistral OCR API directly from the pod
+    (``MISTRAL_API_KEY``), for self-hosters / deployments without the gateway.
+
+``auto`` prefers the gateway (if ``EMBEDDING_GATEWAY_URL`` is set) then direct
+Mistral (if ``MISTRAL_API_KEY``). Both return GitHub-flavoured markdown + exact
+``page_boundaries``; bbox is re-derived from the PDF bytes + boundaries by
+``search/pdf_highlighter``, as for the other tiers.
+"""
+
+import base64
+import logging
+from abc import ABC, abstractmethod
+from collections.abc import Awaitable, Callable
+from typing import Any
+
+import anyio
+import httpx
+
+from nextcloud_mcp_server.config import Settings, get_settings
+
+from .base import DocumentProcessor, ProcessingResult
+
+logger = logging.getLogger(__name__)
+
+_OCR_TIMEOUT_SECONDS = 180.0
+
+
+def _pages_to_text(
+    pages: list[tuple[int, str]],
+) -> tuple[str, list[dict[str, Any]]]:
+    """Join per-page markdown (ordered by index) into one string + boundaries.
+
+    Pages are joined with a blank line. Boundaries are kept CONTIGUOUS (each
+    page owns its leading ``\\n\\n`` separator) so they index exactly into the
+    returned text and ``boundaries[-1]["end_offset"] == len(text)`` -- the
+    ``search/pdf_highlighter`` contract. Consequence: a page's range starts at
+    its separator, not its first glyph (the fast pypdfium2 path joins with no
+    separator, so its ranges are glyph-tight). The 2-char offset is immaterial
+    to page-level chunk attribution.
+    """
+    sep = "\n\n"
+    parts: list[str] = []
+    boundaries: list[dict[str, Any]] = []
+    offset = 0
+    for i, (index, markdown) in enumerate(sorted(pages, key=lambda p: p[0])):
+        chunk = markdown if i == 0 else sep + markdown
+        start = offset
+        offset += len(chunk)
+        parts.append(chunk)
+        boundaries.append(
+            {"page": index + 1, "start_offset": start, "end_offset": offset}
+        )
+    return "".join(parts), boundaries
+
+
+class _OcrBackend(ABC):
+    @abstractmethod
+    async def ocr(
+        self, content: bytes, mime_type: str
+    ) -> tuple[str, list[dict[str, Any]]]: ...
+
+
+class _GatewayOcrBackend(_OcrBackend):
+    """Calls the model gateway's ``POST /v1/ocr`` (key-isolated, M2M-authed)."""
+
+    def __init__(self, base_url: str, model: str, token_provider: Any = None):
+        base = base_url.rstrip("/")
+        if not base.endswith("/v1"):
+            base = f"{base}/v1"
+        self._url = f"{base}/ocr"
+        self._model = model
+        self._token_provider = token_provider
+
+    async def ocr(
+        self, content: bytes, mime_type: str
+    ) -> tuple[str, list[dict[str, Any]]]:
+        headers: dict[str, str] = {}
+        if self._token_provider is not None:
+            headers["Authorization"] = (
+                f"Bearer {await self._token_provider.get_token()}"
+            )
+        payload = {
+            "model": self._model,
+            "document_b64": base64.b64encode(content).decode("ascii"),
+            "mime_type": mime_type,
+        }
+        async with httpx.AsyncClient(
+            timeout=httpx.Timeout(_OCR_TIMEOUT_SECONDS, connect=10.0)
+        ) as client:
+            resp = await client.post(self._url, json=payload, headers=headers)
+            resp.raise_for_status()
+            body = resp.json()
+        pages = [(p["index"], p.get("markdown", "")) for p in body.get("pages", [])]
+        return _pages_to_text(pages)
+
+
+class _MistralOcrBackend(_OcrBackend):
+    """Calls the Mistral OCR API directly (provider key lives in the pod)."""
+
+    def __init__(self, api_key: str, model: str, base_url: str | None = None):
+        from mistralai.client import Mistral  # noqa: PLC0415 -- lazy SDK import
+
+        self._client = Mistral(api_key=api_key, server_url=base_url)
+        # The gateway-namespaced "mistral/<model>" id strips down to the bare
+        # upstream model the SDK expects.
+        self._model = model.split("/", 1)[-1]
+
+    async def ocr(
+        self, content: bytes, mime_type: str
+    ) -> tuple[str, list[dict[str, Any]]]:
+        data_url = (
+            f"data:{mime_type};base64,{base64.b64encode(content).decode('ascii')}"
+        )
+        resp = await self._client.ocr.process_async(
+            model=self._model,
+            document={"type": "document_url", "document_url": data_url},
+        )
+        pages = [(p.index, p.markdown or "") for p in (resp.pages or [])]
+        return _pages_to_text(pages)
+
+
+def build_ocr_backend(settings: Settings) -> _OcrBackend | None:
+    """Select an OCR backend from settings, or None when none is available."""
+    provider = settings.document_ocr_provider
+    if provider == "none":
+        return None
+
+    if provider in ("gateway", "auto") and settings.embedding_gateway_url:
+        token_provider = None
+        if settings.embedding_gateway_client_id:
+            # Lazy import avoids a document_processors -> embedding cycle at load.
+            from ..embedding.gateway_client import (  # noqa: PLC0415
+                GatewayTokenProvider,
+            )
+
+            # Explicit (not assert -- assert is stripped under `python -O`): the
+            # M2M triple is all-or-nothing.
+            if not settings.embedding_gateway_token_url:
+                raise ValueError(
+                    "EMBEDDING_GATEWAY_TOKEN_URL is required when "
+                    "EMBEDDING_GATEWAY_CLIENT_ID is set"
+                )
+            if not settings.embedding_gateway_client_secret:
+                raise ValueError(
+                    "EMBEDDING_GATEWAY_CLIENT_SECRET is required when "
+                    "EMBEDDING_GATEWAY_CLIENT_ID is set"
+                )
+            token_provider = GatewayTokenProvider(
+                token_url=settings.embedding_gateway_token_url,
+                client_id=settings.embedding_gateway_client_id,
+                client_secret=settings.embedding_gateway_client_secret,
+                scope=settings.embedding_gateway_scope,
+            )
+        return _GatewayOcrBackend(
+            settings.embedding_gateway_url, settings.document_ocr_model, token_provider
+        )
+
+    if provider in ("mistral", "auto") and settings.mistral_api_key:
+        return _MistralOcrBackend(
+            settings.mistral_api_key,
+            settings.document_ocr_model,
+            settings.mistral_base_url,
+        )
+
+    # An EXPLICIT provider that's missing its config is an operator error -- warn
+    # loudly (once, since the backend is resolved+cached) rather than silently
+    # disabling OCR. "auto"/"none" fall through to None quietly by design.
+    if provider == "gateway":
+        logger.warning(
+            "DOCUMENT_OCR_PROVIDER=gateway but EMBEDDING_GATEWAY_URL is unset; "
+            "OCR is disabled"
+        )
+    elif provider == "mistral":
+        logger.warning(
+            "DOCUMENT_OCR_PROVIDER=mistral but MISTRAL_API_KEY is unset; "
+            "OCR is disabled"
+        )
+    return None
+
+
+class OcrProcessor(DocumentProcessor):
+    """Tier-3 OCR processor (gateway or direct Mistral backend)."""
+
+    def __init__(self) -> None:
+        # Resolve the backend once and reuse it: rebuilding per call would create
+        # a fresh GatewayTokenProvider each time (discarding its M2M-token cache
+        # -> a token fetch per document) and a new Mistral SDK client per call.
+        # A config change needs a pod restart anyway, so caching for the pod's
+        # lifetime is safe.
+        self._backend_resolved = False
+        self._backend: _OcrBackend | None = None
+        # Serialise first-call resolution so a burst of concurrent OCR requests
+        # doesn't each build a backend (and fetch its own M2M token). Lazy-init:
+        # anyio primitives must not be created at import time.
+        self._backend_lock: anyio.Lock | None = None
+
+    @property
+    def name(self) -> str:
+        return "ocr"
+
+    @property
+    def tier(self) -> str:
+        return "ocr"
+
+    @property
+    def supported_mime_types(self) -> set[str]:
+        return {"application/pdf"}
+
+    async def process(
+        self,
+        content: bytes,
+        content_type: str,
+        filename: str | None = None,
+        options: dict[str, Any] | None = None,
+        progress_callback: (
+            Callable[[float, float | None, str | None], Awaitable[None]] | None
+        ) = None,
+    ) -> ProcessingResult:
+        settings = get_settings()
+        if not self._backend_resolved:
+            if self._backend_lock is None:
+                self._backend_lock = anyio.Lock()
+            async with self._backend_lock:
+                if not self._backend_resolved:  # double-checked
+                    self._backend = build_ocr_backend(settings)
+                    self._backend_resolved = True
+        backend = self._backend
+        if backend is None:
+            logger.warning(
+                "OCR requested for %s but no backend is configured (provider=%s)",
+                filename or "<bytes>",
+                settings.document_ocr_provider,
+            )
+            return ProcessingResult(
+                text="",
+                metadata={"parse_failed_reason": "unsupported"},
+                processor=self.name,
+                success=False,
+                error="no OCR backend configured",
+            )
+        try:
+            text, boundaries = await backend.ocr(
+                content, content_type.split(";")[0].strip().lower()
+            )
+        except Exception as e:
+            logger.warning("OCR failed for %s: %s", filename or "<bytes>", e)
+            return ProcessingResult(
+                text="",
+                metadata={"parse_failed_reason": "error"},
+                processor=self.name,
+                success=False,
+                error=f"{type(e).__name__}: {e}",
+            )
+        return ProcessingResult(
+            text=text,
+            metadata={
+                "page_count": len(boundaries),
+                "page_boundaries": boundaries,
+                "file_size": len(content),
+            },
+            processor=self.name,
+        )
+
+    async def health_check(self) -> bool:
+        # Backends are resolved lazily (and configured per tenant), so there is
+        # nothing to probe here without making a billable upstream call -- the
+        # processor reports healthy and surfaces a real failure per-document.
+        return True

nextcloud_mcp_server/document_processors/pymupdf.py

@@ -73,6 +73,13 @@ class PyMuPDFProcessor(DocumentProcessor):
     def name(self) -> str:
         return "pymupdf"
 
+    @property
+    def tier(self) -> str:
+        # pymupdf4llm recovers markdown structure (headings, lists, tables) via
+        # the expensive graphics-limited table detection -- it is the
+        # ``structured`` escalation target above the pypdfium2 ``fast`` tier.
+        return "structured"
+
     @property
     def supported_mime_types(self) -> set[str]:
         return self.SUPPORTED_TYPES

nextcloud_mcp_server/document_processors/pypdfium2_fast.py

@@ -0,0 +1,128 @@
+"""Tier-1 fast PDF text extractor (pypdfium2).
+
+A permissively-licensed (Apache/BSD-2) fast path that extracts a PDF's text
+layer + page boundaries WITHOUT pymupdf4llm's expensive O(n^2) table/graphics
+analysis. For born-digital PDFs (the tier-0 classifier's ``fast`` verdict) this
+returns clean text in well under a second -- including the form/table PDFs that
+timed out under pymupdf4llm (e.g. ``Student 1a.pdf``: 120s timeout -> ~1s here).
+
+bbox is re-derived from the PDF bytes + ``page_boundaries`` by
+``search/pdf_highlighter``, so this processor only needs to emit ``text`` and
+``metadata["page_boundaries"]`` for chunk highlighting to keep working.
+
+It deliberately does NOT recover tables/layout; a low-quality result is meant to
+escalate to the ``structured`` tier (pymupdf4llm, graphics_limit-guarded) via the
+registry (B2 escalation wiring).
+"""
+
+import logging
+from collections.abc import Awaitable, Callable
+from typing import Any
+
+import anyio
+
+from .base import DocumentProcessor, ProcessingResult
+
+logger = logging.getLogger(__name__)
+
+
+def _extract(content: bytes) -> tuple[str, dict[str, Any]]:
+    """Extract concatenated text + metadata from a PDF (runs in a worker thread).
+
+    ``page_boundaries`` offsets index into the returned text, which is the page
+    texts joined with no separator so the offsets stay exact (the contract
+    ``search/pdf_highlighter`` and the chunker rely on).
+    """
+    import pypdfium2 as pdfium  # noqa: PLC0415 -- keep the native import lazy
+
+    pdf = pdfium.PdfDocument(content)
+    try:
+        page_texts: list[str] = []
+        for i in range(len(pdf)):
+            page = pdf[i]
+            textpage = page.get_textpage()
+            try:
+                page_texts.append(textpage.get_text_bounded() or "")
+            finally:
+                textpage.close()
+                page.close()
+        doc_meta = pdf.get_metadata_dict() or {}
+    finally:
+        pdf.close()
+
+    page_boundaries: list[dict[str, Any]] = []
+    offset = 0
+    for n, text in enumerate(page_texts, start=1):
+        page_boundaries.append(
+            {"page": n, "start_offset": offset, "end_offset": offset + len(text)}
+        )
+        offset += len(text)
+
+    full_text = "".join(page_texts)
+    metadata: dict[str, Any] = {
+        "page_count": len(page_texts),
+        "page_boundaries": page_boundaries,
+    }
+    title = doc_meta.get("Title")
+    if title:
+        metadata["title"] = title
+    return full_text, metadata
+
+
+class Pypdfium2FastProcessor(DocumentProcessor):
+    """Tier-1 fast PDF text extractor backed by pypdfium2."""
+
+    @property
+    def name(self) -> str:
+        return "pypdfium2_fast"
+
+    @property
+    def tier(self) -> str:
+        return "fast"
+
+    @property
+    def supported_mime_types(self) -> set[str]:
+        return {"application/pdf"}
+
+    async def process(
+        self,
+        content: bytes,
+        content_type: str,
+        filename: str | None = None,
+        options: dict[str, Any] | None = None,
+        progress_callback: (
+            Callable[[float, float | None, str | None], Awaitable[None]] | None
+        ) = None,
+    ) -> ProcessingResult:
+        if progress_callback:
+            await progress_callback(0, 100, "Extracting text (pypdfium2)")
+        try:
+            full_text, metadata = await anyio.to_thread.run_sync(  # type: ignore[attr-defined]
+                _extract, content
+            )
+        except Exception as e:
+            # Fast path is best-effort: a failure here escalates rather than
+            # crashing the pipeline. pypdfium2 has no O(n^2) bomb, so this is a
+            # genuinely malformed PDF, not a resource blowup.
+            logger.warning(
+                "pypdfium2 fast extract failed for %s: %s", filename or "<bytes>", e
+            )
+            return ProcessingResult(
+                text="",
+                metadata={"parse_failed_reason": "error"},
+                processor=self.name,
+                success=False,
+                error=f"{type(e).__name__}: {e}",
+            )
+        metadata["file_size"] = len(content)
+        if progress_callback:
+            await progress_callback(100, 100, "Done")
+        return ProcessingResult(text=full_text, metadata=metadata, processor=self.name)
+
+    async def health_check(self) -> bool:
+        try:
+            import pypdfium2  # noqa: F401, PLC0415 -- availability probe
+
+            return True
+        except Exception:
+            return False

nextcloud_mcp_server/document_processors/registry.py

@@ -5,10 +5,16 @@ import time
 from collections.abc import Awaitable, Callable
 from typing import Any
 
-from nextcloud_mcp_server.observability.metrics import record_document_parse
+from nextcloud_mcp_server.config import get_settings
+from nextcloud_mcp_server.observability.metrics import (
+    record_document_classification,
+    record_document_escalation,
+    record_document_parse,
+)
 from nextcloud_mcp_server.observability.tracing import trace_operation
 
 from .base import DocumentProcessor, ProcessingResult, ProcessorError
+from .classifier import classify_from_text
 
 logger = logging.getLogger(__name__)
 
@@ -140,7 +146,7 @@ class ProcessorRegistry:
         Raises:
             ProcessorError: If no processor found or processing fails
         """
-        # Find processor
+        # Forced processor bypasses tiering.
         if processor_name:
             processor = self.get_processor(processor_name)
             if not processor:
@@ -148,14 +154,169 @@ class ProcessorRegistry:
                     f"Processor '{processor_name}' not found. "
                     f"Available: {', '.join(self.list_processors())}"
                 )
-        else:
+            return await self._run_processor(
+                processor, content, content_type, filename, options, progress_callback
+            )
+
+        # PDFs go through the tiered pipeline (tier-0 classify -> tier-1 fast ->
+        # tier-3 OCR escalation). Everything else uses priority selection.
+        if content_type.split(";")[0].strip().lower() == "application/pdf":
+            return await self._process_pdf(
+                content, content_type, filename, options, progress_callback
+            )
+
+        processor = self.find_processor(content_type)
+        if not processor:
+            raise ProcessorError(
+                f"No processor found for type: {content_type}. "
+                f"Registered processors: {', '.join(self.list_processors())}"
+            )
+        return await self._run_processor(
+            processor, content, content_type, filename, options, progress_callback
+        )
+
+    def _pdf_processor_for_tier(self, tier: str) -> DocumentProcessor | None:
+        """First registered processor of ``tier`` that handles PDFs."""
+        for name in self._priority_order:
+            processor = self._processors[name][0]
+            if processor.tier == tier and processor.supports("application/pdf"):
+                return processor
+        return None
+
+    async def _process_pdf(
+        self,
+        content: bytes,
+        content_type: str,
+        filename: str | None,
+        options: dict[str, Any] | None,
+        progress_callback: (
+            Callable[[float, float | None, str | None], Awaitable[None]] | None
+        ),
+    ) -> ProcessingResult:
+        """Tiered PDF pipeline.
+
+        pypdfium2 ``fast`` extracts first; classification is then derived from
+        that text (no PDF re-open), and a scanned/no-text-layer doc escalates to
+        the ``ocr`` tier when enabled. ``document_tier1_engine="pymupdf"`` is a
+        deprecated rollback that pins the structured engine instead.
+        """
+        settings = get_settings()
+
+        if settings.document_tier1_engine == "pymupdf":
+            processor = self._pdf_processor_for_tier("structured")
+            if processor is None:
+                # The rollback was set to opt OUT of pypdfium2, so falling back
+                # to it (the highest-priority PDF processor) silently would
+                # defeat that intent -- warn loudly.
+                processor = self.find_processor(content_type)
+                if processor is None:
+                    raise ProcessorError("No PDF processor registered")
+                logger.warning(
+                    "document_tier1_engine=pymupdf but no 'structured' processor "
+                    "is registered; falling back to '%s'",
+                    processor.name,
+                )
+            return await self._run_processor(
+                processor, content, content_type, filename, options, progress_callback
+            )
+
+        fast = self._pdf_processor_for_tier("fast")
+        if fast is None:
             processor = self.find_processor(content_type)
-            if not processor:
-                raise ProcessorError(
-                    f"No processor found for type: {content_type}. "
-                    f"Registered processors: {', '.join(self.list_processors())}"
+            if processor is None:
+                raise ProcessorError("No PDF processor registered")
+            return await self._run_processor(
+                processor, content, content_type, filename, options, progress_callback
+            )
+
+        result = await self._run_processor(
+            fast, content, content_type, filename, options, progress_callback
+        )
+
+        # Tier-0 classification from the extraction (cheap: no PDF re-open).
+        classification = None
+        if settings.document_classify_enabled and result.success:
+            try:
+                classification = classify_from_text(
+                    result.text, result.metadata.get("page_boundaries") or []
+                )
+                record_document_classification(
+                    classification.recommended_tier,
+                    classification.flags,
+                    classification.mean_text_quality,
+                )
+            except Exception:
+                logger.warning(
+                    "Tier-0 classification failed for %s",
+                    filename or "<bytes>",
+                    exc_info=True,
                 )
 
+        # Escalate scanned / no-text-layer PDFs to OCR (tier-3) when enabled and
+        # a provider is registered. The fast tier is terminal otherwise. Note: a
+        # fast FAILURE (encrypted/corrupt -- result.success False, no
+        # classification) is NOT escalated; a PDF pypdfium2 can't open is treated
+        # as a hard failure (OCR reads the same bytes and would usually fail
+        # too). The page_count guard skips a zero-page (empty/corrupt) PDF, which
+        # OCR can't help either.
+        if (
+            classification is not None
+            and classification.recommended_tier == "ocr"
+            and classification.page_count > 0
+            and settings.document_ocr_enabled
+        ):
+            ocr = self._pdf_processor_for_tier("ocr")
+            if ocr is not None:
+                reason = (
+                    "empty_text"
+                    if classification.total_chars == 0
+                    else "low_confidence"
+                )
+                record_document_escalation("fast", "ocr", reason)
+                logger.info(
+                    "Escalating %s fast->ocr (reason=%s)",
+                    filename or "<bytes>",
+                    reason,
+                )
+                ocr_result = await self._run_processor(
+                    ocr,
+                    content,
+                    content_type,
+                    filename,
+                    options,
+                    progress_callback,
+                    escalated=True,
+                )
+                # OCR is an enhancement, not a gate: if it can't run (no backend
+                # configured / API down) or returns nothing, keep the tier-1
+                # result rather than failing the document. Otherwise an operator
+                # who sets DOCUMENT_OCR_ENABLED=true without credentials would
+                # make scanned docs fail entirely -- strictly worse than off.
+                if ocr_result.success:
+                    return ocr_result
+                logger.warning(
+                    "OCR escalation did not succeed for %s (%s); keeping the "
+                    "tier-1 result",
+                    filename or "<bytes>",
+                    ocr_result.metadata.get("parse_failed_reason", "error"),
+                )
+
+        return result
+
+    async def _run_processor(
+        self,
+        processor: DocumentProcessor,
+        content: bytes,
+        content_type: str,
+        filename: str | None = None,
+        options: dict[str, Any] | None = None,
+        progress_callback: (
+            Callable[[float, float | None, str | None], Awaitable[None]] | None
+        ) = None,
+        *,
+        escalated: bool = False,
+    ) -> ProcessingResult:
+        """Run one processor with the per-processor span + parse metrics."""
         tier = processor.tier
         logger.info(
             "Processing with '%s' processor",
@@ -167,11 +328,6 @@ class ProcessorRegistry:
             },
         )
 
-        # Process (instrumented: per-processor span + parse metrics).
-        # NOTE: when the tiered pipeline (docling/OCR/LLM) lands, escalation
-        # decisions are recorded here via record_document_escalation() and an
-        # add_span_event("document.escalation", ...) -- the escalated=False
-        # attribute and the metric are wired ahead of that.
         byte_size = len(content)
         start_time = time.time()
         with trace_operation(
@@ -181,7 +337,7 @@ class ProcessorRegistry:
                 "processor.tier": tier,
                 "mime_type": content_type,
                 "byte_size": byte_size,
-                "escalated": False,
+                "escalated": escalated,
             },
             record_exception=True,
         ) as span:
@@ -217,6 +373,10 @@ class ProcessorRegistry:
                 raise
 
             duration = time.time() - start_time
+            # Record the tier that actually produced this result so downstream
+            # (Qdrant payload pipeline_tier, analytics) reflects escalation
+            # instead of a hardcoded "fast".
+            result.metadata.setdefault("pipeline_tier", tier)
             pages = int(result.metadata.get("page_count", 0) or 0)
             chars = len(result.text)
             status = "success" if result.success else "error"

nextcloud_mcp_server/vector/processor.py

@@ -17,12 +17,10 @@ from nextcloud_mcp_server.acl_hash import compute_acl_hash
 from nextcloud_mcp_server.client import NextcloudClient
 from nextcloud_mcp_server.config import get_settings
 from nextcloud_mcp_server.document_processors import get_registry
-from nextcloud_mcp_server.document_processors.classifier import classify_pdf
 from nextcloud_mcp_server.embedding import get_bm25_service, get_embedding_service
 from nextcloud_mcp_server.models.deck import DeckCard
 from nextcloud_mcp_server.observability.metrics import (
     record_document_chunks,
-    record_document_classification,
     record_document_parse_failed,
     record_embedding,
     record_qdrant_operation,
@@ -167,36 +165,6 @@ async def processor_task(
     logger.info("Processor %s stopped", worker_id)
 
 
-async def _shadow_classify(content: bytes, content_type: str, file_path: str) -> None:
-    """Tier-0 classification in SHADOW mode: emit metrics, change no routing.
-
-    Best-effort and out of the indexing critical path -- it must never block or
-    fail indexing. PDFs only (the classifier is PDF-specific). The cheap pre-pass
-    runs in a worker thread so it doesn't stall the event loop.
-    """
-    if content_type != "application/pdf":
-        return
-    try:
-        c = await anyio.to_thread.run_sync(classify_pdf, content)  # type: ignore[attr-defined]
-        record_document_classification(c.recommended_tier, c.flags, c.mean_text_quality)
-        logger.debug(
-            "Tier-0 classified %s: tier=%s flags=%s quality=%s",
-            file_path,
-            c.recommended_tier,
-            sorted(c.flags),
-            c.mean_text_quality,
-        )
-    except Exception:
-        # Best-effort: shadow classification must never break indexing, but log
-        # at WARNING (not DEBUG) so a systematic failure -- a pymupdf bug, memory
-        # pressure on every PDF -- stays visible at the production LOG_LEVEL=INFO.
-        logger.warning(
-            "Tier-0 classification failed for %s (shadow mode, indexing unaffected)",
-            file_path,
-            exc_info=True,
-        )
-
-
 async def process_document(
     doc_task: DocumentTask, nc_client: NextcloudClient, *, max_retries: int = 3
 ):
@@ -571,11 +539,8 @@ async def _index_document(
                 "vector_sync.file_size": len(content_bytes),
             },
         ):
-            # Tier-0 shadow classification (observability only; no routing change).
-            if settings.document_classify_enabled:
-                await _shadow_classify(content_bytes, content_type, file_path)
-
-            # Use document processor registry to extract text
+            # The registry runs the tiered PDF pipeline (tier-0 classify ->
+            # tier-1 fast -> OCR escalation) and records classification metrics.
             registry = get_registry()
 
             try:
@@ -632,20 +597,12 @@ async def _index_document(
                 # Diagnostic: Log page boundary information if available
                 if "page_boundaries" in file_metadata:
                     page_boundaries = file_metadata["page_boundaries"]
-                    logger.info(
+                    logger.debug(
                         "Page boundaries for %s: %s pages, text length: %s",
                         file_path,
                         len(page_boundaries),
                         len(content),
                     )
-                    # Log first 3 page boundaries for debugging
-                    for boundary in page_boundaries[:3]:
-                        logger.debug(
-                            "  Page %s: offsets [%s:%s]",
-                            boundary["page"],
-                            boundary["start_offset"],
-                            boundary["end_offset"],
-                        )
                     # Verify last boundary matches text length
                     if page_boundaries:
                         last_boundary = page_boundaries[-1]
@@ -695,23 +652,13 @@ async def _index_document(
 
             # Diagnostic: Verify page number assignment
             assigned_count = sum(1 for c in chunks if c.page_number is not None)
-            logger.info(
+            logger.debug(
                 "Assigned page numbers to %s/%s chunks for %s",
                 assigned_count,
                 len(chunks),
                 file_path,
             )
 
-            # Log first 3 chunks to see their page assignments
-            for i, chunk in enumerate(chunks[:3]):
-                logger.debug(
-                    "  Chunk %s: page=%s, offsets=[%s:%s]",
-                    i,
-                    chunk.page_number,
-                    chunk.start_offset,
-                    chunk.end_offset,
-                )
-
             # Warning if NO page numbers were assigned
             if assigned_count == 0:
                 logger.warning(
@@ -969,7 +916,11 @@ async def _index_document(
                     # Decomposition payload keys (design §10.2), additive.
                     payload_keys.PROCESSOR_VERSION: "monolith-v1",
                     payload_keys.PARSED_AT: indexed_at,
-                    payload_keys.PIPELINE_TIER: "fast",
+                    # Actual tier that produced this doc (registry stamps it on
+                    # the result metadata); non-PDF doc types stay "fast".
+                    payload_keys.PIPELINE_TIER: file_metadata.get(
+                        "pipeline_tier", "fast"
+                    ),
                     payload_keys.EMBEDDING_IDENTITY: _embedding_identity,
                     payload_keys.ACL_HASH: _acl_hash,
                     # File-specific metadata (PDF, etc.)