feat: tier-0 document classifier in shadow mode

First step of the tiered document-processor effort (Deck #203): a cheap, local
pre-pass that recommends which extraction tier a PDF should start in, emitting
metrics WITHOUT changing routing yet -- so we gather per-tenant doc-mix data
before turning escalation on.

document_processors/classifier.py: classify_pdf(content) -> DocClassification.
Page-sampled (bounded on large docs), <~1s. Cheap signals only -- text-layer
chars, a text-quality score (catches the "Student 147" failure where a text
layer exists but is mashed/space-less junk), and image coverage. A page that is
mostly a raster image routes to OCR: its content (handwriting, stamps) isn't in
any text layer. Deliberately no get_drawings/graphics-density signal -- it's
slow on the exact pages it'd flag, the hotfix's graphics_limit already makes the
parse safe, and the (future) tier-1 quality gate catches lost tables.

Validated on the sample corpus: born-digital 2-col arxiv and a digital student
record -> fast (tier 1); a scanned+handwritten form -> ocr (tier 3).

Wiring (vector/processor.py): _shadow_classify runs the classifier on PDFs in a
worker thread, best-effort (never blocks/fails indexing), gated by the new
DOCUMENT_CLASSIFY_ENABLED setting. Metrics: astrolabe_document_classified_total
{recommended_tier}, astrolabe_document_classifier_flag_total{flag},
astrolabe_document_text_quality histogram.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

nextcloud_mcp_server/config.py

@@ -136,6 +136,8 @@ _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)
+    "document_classify_enabled": True,
     # Observability
     "metrics_enabled": True,
     "metrics_port": 9090,
@@ -727,6 +729,9 @@ 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.
+    document_classify_enabled: bool = True
 
     # Observability settings
     metrics_enabled: bool = True
@@ -1339,6 +1344,7 @@ def get_settings() -> Settings:
         "document_pdf_graphics_limit": "DOCUMENT_PDF_GRAPHICS_LIMIT",
         "document_parse_timeout_seconds": "DOCUMENT_PARSE_TIMEOUT_SECONDS",
         "document_parse_mem_limit_mb": "DOCUMENT_PARSE_MEM_LIMIT_MB",
+        "document_classify_enabled": "DOCUMENT_CLASSIFY_ENABLED",
         # Observability settings
         "metrics_enabled": "METRICS_ENABLED",
         "metrics_port": "METRICS_PORT",

nextcloud_mcp_server/document_processors/classifier.py

@@ -0,0 +1,169 @@
+"""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.
+
+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):
+
+  * 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)
+
+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.
+"""
+
+import logging
+import re
+from dataclasses import dataclass, field
+
+logger = logging.getLogger(__name__)
+
+# Page-sampling: classify at most this many pages on large docs (evenly spaced)
+# so the pass stays bounded regardless of page count.
+MAX_SAMPLED_PAGES = 24
+
+# A page counts as "scanned-like" when a raster image covers most of it.
+IMAGE_COVERAGE_SCANNED = 0.80
+# Text-quality score below which the layer is treated as junk (mashed tokens).
+MIN_TEXT_QUALITY = 0.45
+# Fraction of sampled pages that must look scanned/bad for a doc->ocr verdict.
+OCR_PAGE_FRACTION = 0.5
+
+_WORD_RE = re.compile(r"\S+")
+
+
+@dataclass
+class PageSignals:
+    page_no: int
+    char_count: int
+    image_coverage: float  # 0..1 of page area covered by images
+    text_quality: float  # 0..1; low = mashed/space-less/garbage layer
+    needs_ocr: bool  # scanned or unusable text layer
+
+
+@dataclass
+class DocClassification:
+    page_count: int
+    sampled_pages: int
+    total_chars: int
+    mean_text_quality: float
+    ocr_page_fraction: float  # fraction of sampled pages flagged needs_ocr
+    recommended_tier: str  # "fast" | "ocr"
+    flags: set[str] = field(
+        default_factory=set
+    )  # scanned | bad_text_layer | image_heavy
+    pages: list[PageSignals] = field(default_factory=list)
+
+
+def _text_quality(text: str) -> float:
+    """Score a text layer's usability in ``[0, 1]`` (1 = clean prose).
+
+    Penalises the two hallmarks of a junk/OCR-mangled layer: too little
+    whitespace (words mashed together) and very long tokens. Empty text scores
+    0 -- "no usable layer".
+    """
+    if not text:
+        return 0.0
+    tokens = _WORD_RE.findall(text)
+    if not tokens:
+        return 0.0
+    whitespace_ratio = sum(c.isspace() for c in text) / len(text)
+    mean_token_len = sum(len(t) for t in tokens) / len(tokens)
+    overlong_frac = sum(len(t) > 20 for t in tokens) / len(tokens)
+    # Clean prose: ~15-20% whitespace, mean token ~4-6 chars, ~no overlong tokens.
+    ws_score = min(whitespace_ratio / 0.12, 1.0)
+    len_score = (
+        1.0 if mean_token_len <= 10 else max(0.0, 1.0 - (mean_token_len - 10) / 15)
+    )
+    overlong_score = max(0.0, 1.0 - overlong_frac * 5)
+    return round(ws_score * len_score * overlong_score, 3)
+
+
+def _sample_indices(page_count: int) -> list[int]:
+    if page_count <= MAX_SAMPLED_PAGES:
+        return list(range(page_count))
+    # Evenly spaced sample across the document.
+    step = page_count / MAX_SAMPLED_PAGES
+    return sorted({int(i * step) for i in range(MAX_SAMPLED_PAGES)})
+
+
+def classify_pdf(content: bytes) -> DocClassification:
+    """Classify a PDF from its bytes.
+
+    May raise (e.g. ``pymupdf`` errors) if the bytes can't be opened as a PDF;
+    callers run it in a guarded context (shadow mode swallows failures) so a
+    bad file never breaks indexing.
+    """
+    import pymupdf  # noqa: PLC0415 -- keep the heavy import lazy / off module load
+
+    doc = pymupdf.open("pdf", content)
+    try:
+        page_count = doc.page_count
+        indices = _sample_indices(page_count)
+        pages: list[PageSignals] = []
+        for n in indices:
+            page = doc.load_page(n)
+            text = page.get_text("text")
+            quality = _text_quality(text)
+            page_area = abs(page.rect.width * page.rect.height) or 1.0
+            img_area = 0.0
+            for img in page.get_images(full=True):
+                for rect in page.get_image_rects(img[0]):
+                    img_area += abs(rect.width * rect.height)
+            coverage = min(img_area / page_area, 1.0)
+            # A page that is mostly a raster image is a scan/photo: its content
+            # (handwriting, stamps, figure text) is not fully in any text layer,
+            # so OCR is needed to capture it -- regardless of whether a partial
+            # text layer is present. Text quality/char-count are kept as
+            # diagnostic signals (flags + tuning metrics), not the trigger,
+            # because OCR only helps when there is an image to read.
+            needs_ocr = coverage >= IMAGE_COVERAGE_SCANNED
+            pages.append(
+                PageSignals(n, len(text), round(coverage, 3), quality, needs_ocr)
+            )
+    finally:
+        doc.close()
+
+    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
+    )
+    ocr_frac = (sum(p.needs_ocr for p in pages) / sampled) if sampled else 0.0
+
+    flags: set[str] = set()
+    if any(p.image_coverage >= IMAGE_COVERAGE_SCANNED for p in pages):
+        flags.add("image_heavy")
+    if (
+        ocr_frac >= OCR_PAGE_FRACTION
+        and total_chars
+        and mean_quality < MIN_TEXT_QUALITY
+    ):
+        flags.add("bad_text_layer")
+    if ocr_frac >= OCR_PAGE_FRACTION and total_chars == 0:
+        flags.add("scanned")
+
+    recommended = "ocr" if ocr_frac >= OCR_PAGE_FRACTION else "fast"
+
+    return DocClassification(
+        page_count=page_count,
+        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/observability/metrics.py

@@ -272,6 +272,30 @@ document_parse_failed_total = Counter(
     ["reason"],  # reason: timeout | oom | error
 )
 
+# --- Tier-0 classifier (shadow mode) -----------------------------------------
+#
+# The classifier runs a cheap pre-pass per PDF and recommends a starting tier.
+# In shadow mode it changes no routing -- these metrics gather the per-tenant
+# doc-mix needed to tune the thresholds before routing is enabled.
+
+document_classified_total = Counter(
+    "astrolabe_document_classified_total",
+    "Documents classified by tier-0, by recommended starting tier",
+    ["recommended_tier"],  # fast | ocr
+)
+
+document_classifier_flag_total = Counter(
+    "astrolabe_document_classifier_flag_total",
+    "Tier-0 classifier flags raised on documents",
+    ["flag"],  # image_heavy | scanned | bad_text_layer
+)
+
+document_text_quality = Histogram(
+    "astrolabe_document_text_quality",
+    "Tier-0 mean text-layer quality per document (0=junk, 1=clean prose)",
+    buckets=(0.1, 0.2, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8, 0.9, 1.0),
+)
+
 # --- Embedding stages ---------------------------------------------------------
 
 embedding_duration_seconds = Histogram(
@@ -637,6 +661,20 @@ def record_document_parse_failed(reason: str) -> None:
     document_parse_failed_total.labels(reason=reason).inc()
 
 
+def record_document_classification(
+    recommended_tier: str, flags: set[str], mean_text_quality: float
+) -> None:
+    """Record a tier-0 classification result (shadow mode -- observability only).
+
+    Primitive args (not the DocClassification object) keep the observability
+    layer free of a dependency on document_processors.
+    """
+    document_classified_total.labels(recommended_tier=recommended_tier).inc()
+    for flag in flags:
+        document_classifier_flag_total.labels(flag=flag).inc()
+    document_text_quality.observe(mean_text_quality)
+
+
 def record_embedding(
     kind: str,
     provider: str,

nextcloud_mcp_server/vector/processor.py

@@ -17,10 +17,12 @@ 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,
@@ -164,6 +166,29 @@ 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:
+        logger.debug("Tier-0 classification failed for %s", file_path, exc_info=True)
+
+
 async def process_document(
     doc_task: DocumentTask, nc_client: NextcloudClient, *, max_retries: int = 3
 ):
@@ -534,6 +559,10 @@ 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
             registry = get_registry()
 

tests/unit/test_doc_classifier.py

@@ -0,0 +1,86 @@
+"""Unit tests for the tier-0 document classifier.
+
+Pins the routing decisions and the text-quality heuristic that drive which
+extraction tier a PDF starts in:
+  * a clean born-digital PDF (text, no full-page images) -> ``fast`` (tier 1);
+  * a full-page-image scan -> ``ocr`` (tier 3), since handwriting/stamps aren't
+    in any text layer;
+  * the text-quality score distinguishes clean prose from mashed/space-less junk.
+"""
+
+import pymupdf
+import pytest
+
+from nextcloud_mcp_server.document_processors import classifier as clf
+
+pytestmark = pytest.mark.unit
+
+
+def _digital_pdf(
+    pages: int = 3, body: str = "Hello world this is clean text. "
+) -> bytes:
+    doc = pymupdf.open()
+    for _ in range(pages):
+        page = doc.new_page(width=595, height=842)
+        page.insert_text((50, 60), body * 8)
+    data: bytes = doc.tobytes()
+    doc.close()
+    return data
+
+
+def _full_page_image_pdf(pages: int = 2) -> bytes:
+    # A page whose entire area is a raster image -> looks scanned.
+    doc = pymupdf.open()
+    pix = pymupdf.Pixmap(pymupdf.csRGB, pymupdf.IRect(0, 0, 600, 850))
+    pix.clear_with(255)
+    img = pix.tobytes("png")
+    for _ in range(pages):
+        page = doc.new_page(width=595, height=842)
+        page.insert_image(page.rect, stream=img)
+    data: bytes = doc.tobytes()
+    doc.close()
+    return data
+
+
+# --- text-quality heuristic --------------------------------------------------
+
+
+def test_text_quality_clean_prose_scores_high():
+    assert clf._text_quality("the quick brown fox jumps over the lazy dog") > 0.8
+
+
+def test_text_quality_mashed_tokens_scores_low():
+    # space-less / mashed layer (the "Student 147" failure mode)
+    mashed = "01322234567mobileoutstandingresilienceacademicachievementhurdles"
+    assert clf._text_quality(mashed) < clf.MIN_TEXT_QUALITY
+
+
+def test_text_quality_empty_is_zero():
+    assert clf._text_quality("") == 0.0
+
+
+# --- routing -----------------------------------------------------------------
+
+
+def test_digital_pdf_routes_fast():
+    c = clf.classify_pdf(_digital_pdf())
+    assert c.recommended_tier == "fast"
+    assert c.ocr_page_fraction == 0.0
+    assert "image_heavy" not in c.flags
+    assert c.mean_text_quality > 0.8
+
+
+def test_full_page_image_routes_ocr():
+    c = clf.classify_pdf(_full_page_image_pdf())
+    assert c.recommended_tier == "ocr"
+    assert c.ocr_page_fraction == 1.0
+    assert "image_heavy" in c.flags
+
+
+# --- sampling bounds large docs ----------------------------------------------
+
+
+def test_large_doc_is_sampled():
+    c = clf.classify_pdf(_digital_pdf(pages=120))
+    assert c.page_count == 120
+    assert c.sampled_pages <= clf.MAX_SAMPLED_PAGES

tests/unit/test_shadow_classify.py

@@ -0,0 +1,62 @@
+"""Tests for the tier-0 shadow-classification wiring in the processor.
+
+Shadow mode = observability only: it emits classification metrics but must never
+block or fail indexing, and only applies to PDFs.
+"""
+
+from unittest.mock import MagicMock
+
+import pytest
+
+from nextcloud_mcp_server.document_processors.classifier import DocClassification
+from nextcloud_mcp_server.vector import processor as proc
+
+pytestmark = pytest.mark.unit
+
+
+def _classification() -> DocClassification:
+    return DocClassification(
+        page_count=2,
+        sampled_pages=2,
+        total_chars=100,
+        mean_text_quality=0.9,
+        ocr_page_fraction=0.0,
+        recommended_tier="fast",
+        flags={"image_heavy"},
+    )
+
+
+async def test_shadow_classify_records_metrics(monkeypatch):
+    monkeypatch.setattr(proc, "classify_pdf", lambda content: _classification())
+    rec = MagicMock()
+    monkeypatch.setattr(proc, "record_document_classification", rec)
+
+    await proc._shadow_classify(b"%PDF-1.7", "application/pdf", "f.pdf")
+
+    rec.assert_called_once_with("fast", {"image_heavy"}, 0.9)
+
+
+async def test_shadow_classify_skips_non_pdf(monkeypatch):
+    called = MagicMock()
+    monkeypatch.setattr(proc, "classify_pdf", called)
+    rec = MagicMock()
+    monkeypatch.setattr(proc, "record_document_classification", rec)
+
+    await proc._shadow_classify(b"plain", "text/plain", "f.txt")
+
+    called.assert_not_called()
+    rec.assert_not_called()
+
+
+async def test_shadow_classify_swallows_errors(monkeypatch):
+    def boom(content):
+        raise ValueError("bad pdf")
+
+    monkeypatch.setattr(proc, "classify_pdf", boom)
+    rec = MagicMock()
+    monkeypatch.setattr(proc, "record_document_classification", rec)
+
+    # Must not raise -- shadow classification is best-effort, off the index path.
+    await proc._shadow_classify(b"%PDF-1.7", "application/pdf", "f.pdf")
+
+    rec.assert_not_called()