diff --git a/nextcloud_mcp_server/document_processors/escalation.py b/nextcloud_mcp_server/document_processors/escalation.py index 7e39d765..d85733f6 100644 --- a/nextcloud_mcp_server/document_processors/escalation.py +++ b/nextcloud_mcp_server/document_processors/escalation.py @@ -21,11 +21,36 @@ mapping lives in the queue layer, which imports :class:`EscalateError` from here from __future__ import annotations +from dataclasses import dataclass + # Cheapest-first. ``llm`` is reserved (see base.DocumentProcessor.tier) and not # wired yet, so it is intentionally absent from the live ladder. TIER_LADDER: tuple[str, ...] = ("fast", "structured", "ocr") +@dataclass(frozen=True) +class EscalationDecision: + """Outcome of the post-parse quality gate (``ProcessorRegistry.evaluate_escalation``). + + ``kind``: + * ``"hop"`` — the parse is too poor and a higher tier *can run*; the caller + raises :class:`EscalateError` to requeue the document onto ``to_tier``. + * ``"suppressed"`` — the parse would escalate to ``to_tier`` (the *ideal* + next tier), but that tier is **disabled** (e.g. OCR off). The caller does + NOT hop — it indexes the current tier's output as terminal — and records + the would-be escalation so operators see the latent demand ("what-if OCR + were enabled"). Enabling the tier turns these into real ``"hop"`` events. + + A ``None`` return from ``evaluate_escalation`` (not an instance of this class) + means "index as-is, nothing to escalate" — good text, or no higher tier + exists at all (no processor registered for it). + """ + + kind: str # "hop" | "suppressed" + to_tier: str + reason: str # empty_text | low_confidence + + def next_tier(current: str) -> str | None: """The next tier above ``current`` in the ladder, or ``None`` if terminal. diff --git a/nextcloud_mcp_server/document_processors/registry.py b/nextcloud_mcp_server/document_processors/registry.py index ff4b678e..87f9dce4 100644 --- a/nextcloud_mcp_server/document_processors/registry.py +++ b/nextcloud_mcp_server/document_processors/registry.py @@ -15,7 +15,7 @@ from nextcloud_mcp_server.observability.tracing import trace_operation from .base import DocumentProcessor, ProcessingResult, ProcessorError from .classifier import DocClassification, classify_from_text, image_coverage_per_page -from .escalation import TIER_LADDER +from .escalation import TIER_LADDER, EscalationDecision logger = logging.getLogger(__name__) @@ -390,22 +390,34 @@ class ProcessorRegistry: ) return classification - def _tier_available(self, tier: str, settings: Any) -> bool: - """Whether ``tier`` can actually run a PDF parse right now. + def _tier_available( + self, tier: str, settings: Any, *, ignore_enabled: bool = False + ) -> bool: + """Whether ``tier`` can run a PDF parse right now. A tier is available when it has a registered PDF processor and is enabled; the ``ocr`` tier additionally requires ``DOCUMENT_OCR_ENABLED`` (so OCR stays opt-in and a misconfigured tenant never escalates to a backend it hasn't turned on). + + ``ignore_enabled`` drops only the *enabled* gate (not the registered- + processor requirement): it answers "would this tier run if it were turned + on?" — used to compute the *ideal* escalation target for the + what-if-OCR suppressed-escalation signal. """ if self._pdf_processor_for_tier(tier) is None: return False - if tier == "ocr" and not settings.document_ocr_enabled: + if not ignore_enabled and tier == "ocr" and not settings.document_ocr_enabled: return False return True def next_available_tier( - self, current_tier: str, settings: Any, *, minimum: str | None = None + self, + current_tier: str, + settings: Any, + *, + minimum: str | None = None, + ignore_enabled: bool = False, ) -> str | None: """First escalation target above ``current_tier`` that can actually run. @@ -413,6 +425,8 @@ class ProcessorRegistry: ``minimum``'s rung, when given) and returns the first :meth:`_tier_available` tier. ``None`` means no higher tier can run -- ``current_tier`` is then terminal and its result is indexed as-is. + ``ignore_enabled`` is forwarded to :meth:`_tier_available` to find the + *ideal* target ignoring the OCR-enabled gate (see ``evaluate_escalation``). """ try: cur_idx = TIER_LADDER.index(current_tier) @@ -425,7 +439,7 @@ class ProcessorRegistry: except ValueError: pass for tier in TIER_LADDER[start_idx:]: - if self._tier_available(tier, settings): + if self._tier_available(tier, settings, ignore_enabled=ignore_enabled): return tier return None @@ -475,25 +489,33 @@ class ProcessorRegistry: settings: Any, *, filename: str | None = None, - ) -> tuple[str, str] | None: + ) -> EscalationDecision | None: """Decide whether ``current_tier``'s result must escalate (external path). - Returns ``(to_tier, reason)`` when the parse is too poor to index and a - higher tier can run, else ``None`` (index the result as-is). Reuses the - tier-0 classifier as the post-parse quality gate, so the escalation - signal is identical to the inline pipeline's. + Returns an :class:`EscalationDecision` (``"hop"`` or ``"suppressed"``) + when the classifier judges the parse too poor to index, else ``None`` + (index as-is). Reuses the tier-0 classifier as the post-parse quality + gate, so the signal is identical to the inline pipeline's. A hard parse FAILURE (``result.success`` False) is never escalated: a corrupt/encrypted PDF one engine can't open usually defeats the others too (OCR reads the same bytes), so the caller marks it failed instead. - Routing of the target tier: + Target-tier routing: - ``total_chars == 0`` (scanned / no text layer) -> target the ``ocr`` tier directly. Text-extractor tiers (``structured``) cannot conjure text from a pure raster scan, so a structured hop would just be wasted. - low-confidence but non-empty layer -> escalate to the next rung, so a different in-cluster extractor can try before paying for OCR. + + Hop vs suppressed: if the ideal target tier can run, return a ``"hop"``. + If it can't run **only because it's disabled** (OCR off — the *ideal* + tier exists ignoring the enabled gate, but the *available* one does not), + return ``"suppressed"`` so the caller records the would-be hop and indexes + the current tier's output as terminal (OCR stays opt-in + cost-free, but + the latent demand is observable). If no higher tier exists *at all* (no + processor registered), it's genuinely terminal -> ``None``. """ classification = self._classify_result( result, @@ -508,14 +530,22 @@ class ProcessorRegistry: if classification.page_count <= 0: return None if classification.total_chars == 0: - to_tier = self.next_available_tier(current_tier, settings, minimum="ocr") + minimum: str | None = "ocr" reason = "empty_text" else: - to_tier = self.next_available_tier(current_tier, settings) + minimum = None reason = "low_confidence" - if to_tier is None: - return None - return (to_tier, reason) + to_tier = self.next_available_tier(current_tier, settings, minimum=minimum) + if to_tier is not None: + return EscalationDecision("hop", to_tier, reason) + # No tier can run as configured. Distinguish "disabled (e.g. OCR off)" + # from "no such tier at all" by re-resolving ignoring the enabled gate. + ideal = self.next_available_tier( + current_tier, settings, minimum=minimum, ignore_enabled=True + ) + if ideal is not None: + return EscalationDecision("suppressed", ideal, reason) + return None async def _run_processor( self, diff --git a/nextcloud_mcp_server/observability/metrics.py b/nextcloud_mcp_server/observability/metrics.py index 74dab2a5..92e87871 100644 --- a/nextcloud_mcp_server/observability/metrics.py +++ b/nextcloud_mcp_server/observability/metrics.py @@ -276,6 +276,21 @@ document_escalation_total = Counter( ["from_tier", "to_tier", "reason"], ) +# Would-be escalations SUPPRESSED because the target tier is disabled (Deck +# #324). The cost-sensitive ``ocr`` tier is opt-in (DOCUMENT_OCR_ENABLED): when +# it's off, a doc the classifier would route to OCR is indexed at the pre-OCR +# tier instead of hopping, and that intent is counted here rather than on +# document_escalation_total. This is the "what-if OCR were enabled" signal — +# escalation_suppressed_total{to_tier="ocr"} is the latent OCR demand an operator +# weighs before enabling OCR; enabling it converts these into real +# document_escalation_total{to_tier="ocr"} hops. +document_escalation_suppressed_total = Counter( + "astrolabe_document_escalation_suppressed_total", + "Would-be tier escalations suppressed because the target tier is disabled", + # reason: low_confidence | empty_text + ["from_tier", "to_tier", "reason"], +) + # Hard parse failures: the parse now runs in an isolated subprocess, so a # timeout/OOM that kills the worker is caught here. This is distinct from # ``document_parse_total{status="error"}`` (an in-process exception): a hard @@ -746,6 +761,20 @@ def record_document_escalation(from_tier: str, to_tier: str, reason: str) -> Non ).inc() +def record_document_escalation_suppressed( + from_tier: str, to_tier: str, reason: str +) -> None: + """Record a would-be escalation suppressed because ``to_tier`` is disabled. + + The "what-if OCR were enabled" signal (Deck #324): the document is indexed at + ``from_tier`` (terminal) rather than hopped, because the ideal next tier + (typically ``ocr``) is turned off. See ``document_escalation_suppressed_total``. + """ + document_escalation_suppressed_total.labels( + from_tier=from_tier, to_tier=to_tier, reason=reason + ).inc() + + def record_document_parse_failed(reason: str) -> None: """Record a hard parse failure from the isolated worker. diff --git a/nextcloud_mcp_server/vector/processor.py b/nextcloud_mcp_server/vector/processor.py index 05619b18..0423550e 100644 --- a/nextcloud_mcp_server/vector/processor.py +++ b/nextcloud_mcp_server/vector/processor.py @@ -28,6 +28,7 @@ from nextcloud_mcp_server.models.deck import DeckCard from nextcloud_mcp_server.observability.metrics import ( record_document_chunks, record_document_escalation, + record_document_escalation_suppressed, record_document_parse_failed, record_embedding, record_embedding_tokens, @@ -156,17 +157,35 @@ async def _parse_pdf_tier( decision = registry.evaluate_escalation( result, content, tier, settings, filename=filename ) - if decision is not None: - to_tier, reason = decision - record_document_escalation(tier, to_tier, reason) + if decision is not None and decision.kind == "suppressed": + # The ideal next tier (e.g. ocr) is disabled, so we do NOT hop: index + # this tier's output as terminal and record the would-be escalation + # so operators see the latent demand ("what-if OCR enabled"; #324). + record_document_escalation_suppressed( + tier, decision.to_tier, decision.reason + ) + logger.info( + "Escalation suppressed for %s %s->%s (reason=%s; %s disabled), " + "indexing at %s", + filename or "", + tier, + decision.to_tier, + decision.reason, + decision.to_tier, + tier, + ) + elif decision is not None: + record_document_escalation(tier, decision.to_tier, decision.reason) logger.info( "Escalating %s %s->%s (reason=%s)", filename or "", tier, - to_tier, - reason, + decision.to_tier, + decision.reason, + ) + raise EscalateError( + from_tier=tier, to_tier=decision.to_tier, reason=decision.reason ) - raise EscalateError(from_tier=tier, to_tier=to_tier, reason=reason) return result diff --git a/tests/unit/test_registry_tiering.py b/tests/unit/test_registry_tiering.py index 7c343b14..0fa38a82 100644 --- a/tests/unit/test_registry_tiering.py +++ b/tests/unit/test_registry_tiering.py @@ -13,6 +13,7 @@ from nextcloud_mcp_server.document_processors.base import ( DocumentProcessor, ProcessingResult, ) +from nextcloud_mcp_server.document_processors.escalation import EscalationDecision from nextcloud_mcp_server.document_processors.registry import ProcessorRegistry pytestmark = pytest.mark.unit @@ -334,7 +335,7 @@ def test_evaluate_escalation_empty_jumps_to_ocr(monkeypatch): processor="fast", ) decision = r.evaluate_escalation(res, b"%PDF", "fast", _Settings(ocr=True)) - assert decision == ("ocr", "empty_text") + assert decision == EscalationDecision("hop", "ocr", "empty_text") def test_evaluate_escalation_lowconf_goes_to_structured(monkeypatch): @@ -357,7 +358,7 @@ def test_evaluate_escalation_lowconf_goes_to_structured(monkeypatch): processor="fast", ) decision = r.evaluate_escalation(res, b"%PDF", "fast", _Settings(ocr=True)) - assert decision == ("structured", "low_confidence") + assert decision == EscalationDecision("hop", "structured", "low_confidence") def test_evaluate_escalation_failure_not_escalated(monkeypatch): @@ -419,4 +420,65 @@ def test_evaluate_escalation_lowconf_to_ocr_when_no_structured(monkeypatch): processor="fast", ) decision = r.evaluate_escalation(res, b"%PDF", "fast", _Settings(ocr=True)) - assert decision == ("ocr", "low_confidence") + assert decision == EscalationDecision("hop", "ocr", "low_confidence") + + +def test_evaluate_escalation_suppressed_when_ocr_disabled(monkeypatch): + """OCR off: a scanned doc does NOT hop to ocr; it returns a 'suppressed' + decision (the what-if-OCR signal) so the caller indexes at the current tier.""" + monkeypatch.setattr(reg_mod, "record_document_classification", MagicMock()) + r = _registry((_Fake("fast", "fast"), 20), (_Fake("ocr", "ocr"), 5)) + res = ProcessingResult( + text="", + metadata={ + "page_count": 1, + "page_boundaries": [{"page": 1, "start_offset": 0, "end_offset": 0}], + }, + processor="fast", + ) + decision = r.evaluate_escalation(res, b"%PDF", "fast", _Settings(ocr=False)) + assert decision == EscalationDecision("suppressed", "ocr", "empty_text") + + +def test_evaluate_escalation_lowconf_suppressed_when_only_ocr_disabled(monkeypatch): + """fast+ocr only, OCR off, junk text: the next rung is the disabled ocr, so + the would-be hop is suppressed (not a structured hop, which isn't registered).""" + monkeypatch.setattr(reg_mod, "record_document_classification", MagicMock()) + junk = "q" * 40 + r = _registry((_Fake("fast", "fast"), 20), (_Fake("ocr", "ocr"), 5)) + res = ProcessingResult( + text=junk, + metadata={ + "page_count": 1, + "page_boundaries": [ + {"page": 1, "start_offset": 0, "end_offset": len(junk)} + ], + }, + processor="fast", + ) + decision = r.evaluate_escalation(res, b"%PDF", "fast", _Settings(ocr=False)) + assert decision == EscalationDecision("suppressed", "ocr", "low_confidence") + + +def test_evaluate_escalation_structured_hop_not_suppressed_when_ocr_off(monkeypatch): + """OCR off but structured available + junk text → real hop to structured + (not suppressed): the in-cluster rung can still run.""" + monkeypatch.setattr(reg_mod, "record_document_classification", MagicMock()) + junk = "w" * 40 + r = _registry( + (_Fake("fast", "fast"), 20), + (_Fake("structured", "structured"), 10), + (_Fake("ocr", "ocr"), 5), + ) + res = ProcessingResult( + text=junk, + metadata={ + "page_count": 1, + "page_boundaries": [ + {"page": 1, "start_offset": 0, "end_offset": len(junk)} + ], + }, + processor="fast", + ) + decision = r.evaluate_escalation(res, b"%PDF", "fast", _Settings(ocr=False)) + assert decision == EscalationDecision("hop", "structured", "low_confidence") diff --git a/tests/unit/vector/test_parse_pdf_tier.py b/tests/unit/vector/test_parse_pdf_tier.py index cc445ea3..48e90b26 100644 --- a/tests/unit/vector/test_parse_pdf_tier.py +++ b/tests/unit/vector/test_parse_pdf_tier.py @@ -1,8 +1,9 @@ -"""Unit tests for the per-tier PDF parse + escalation gate (Deck #323). +"""Unit tests for the per-tier PDF parse + escalation gate (Deck #323/#324). -``processor._parse_pdf_tier`` runs one tier and either returns the result to -index or raises ``EscalateError`` (a queue-hop). These exercise the decision -without standing up the full ingest pipeline. +``processor._parse_pdf_tier`` runs one tier and either indexes the result, +raises ``EscalateError`` (a real queue-hop), or — when the ideal next tier is +disabled (OCR off) — records a suppressed escalation and indexes as terminal. +These exercise the decision without standing up the full ingest pipeline. """ from unittest.mock import AsyncMock, MagicMock @@ -10,7 +11,10 @@ from unittest.mock import AsyncMock, MagicMock import pytest from nextcloud_mcp_server.document_processors.base import ProcessingResult -from nextcloud_mcp_server.document_processors.escalation import EscalateError +from nextcloud_mcp_server.document_processors.escalation import ( + EscalateError, + EscalationDecision, +) from nextcloud_mcp_server.vector import processor pytestmark = pytest.mark.unit @@ -25,7 +29,9 @@ def _registry(result: ProcessingResult, decision): async def test_good_parse_returns_result(monkeypatch): rec = MagicMock() + sup = MagicMock() monkeypatch.setattr(processor, "record_document_escalation", rec) + monkeypatch.setattr(processor, "record_document_escalation_suppressed", sup) result = ProcessingResult(text="clean", metadata={}, processor="fast") reg = _registry(result, decision=None) out = await processor._parse_pdf_tier( @@ -33,13 +39,14 @@ async def test_good_parse_returns_result(monkeypatch): ) assert out is result rec.assert_not_called() + sup.assert_not_called() async def test_low_quality_parse_raises_escalate(monkeypatch): rec = MagicMock() monkeypatch.setattr(processor, "record_document_escalation", rec) result = ProcessingResult(text="", metadata={}, processor="fast") - reg = _registry(result, decision=("ocr", "empty_text")) + reg = _registry(result, decision=EscalationDecision("hop", "ocr", "empty_text")) with pytest.raises(EscalateError) as ei: await processor._parse_pdf_tier( reg, b"%PDF", "application/pdf", "f.pdf", "fast", settings=object() @@ -51,16 +58,37 @@ async def test_low_quality_parse_raises_escalate(monkeypatch): rec.assert_called_once_with("fast", "ocr", "empty_text") +async def test_suppressed_decision_indexes_without_hop(monkeypatch): + """OCR-off (suppressed): index this tier's result, record the would-be hop, + do NOT raise EscalateError.""" + rec = MagicMock() + sup = MagicMock() + monkeypatch.setattr(processor, "record_document_escalation", rec) + monkeypatch.setattr(processor, "record_document_escalation_suppressed", sup) + result = ProcessingResult(text="junk", metadata={}, processor="fast") + reg = _registry( + result, decision=EscalationDecision("suppressed", "ocr", "empty_text") + ) + out = await processor._parse_pdf_tier( + reg, b"%PDF", "application/pdf", "f.pdf", "fast", settings=object() + ) + assert out is result # indexed as terminal, no hop + sup.assert_called_once_with("fast", "ocr", "empty_text") + rec.assert_not_called() + + async def test_hard_failure_returns_result_without_escalating(monkeypatch): rec = MagicMock() + sup = MagicMock() monkeypatch.setattr(processor, "record_document_escalation", rec) + monkeypatch.setattr(processor, "record_document_escalation_suppressed", sup) result = ProcessingResult( text="", metadata={"parse_failed_reason": "oversize"}, processor="size_guard", success=False, ) - reg = _registry(result, decision=("ocr", "empty_text")) + reg = _registry(result, decision=EscalationDecision("hop", "ocr", "empty_text")) out = await processor._parse_pdf_tier( reg, b"%PDF", "application/pdf", "big.pdf", "fast", settings=object() ) @@ -68,3 +96,4 @@ async def test_hard_failure_returns_result_without_escalating(monkeypatch): assert out is result reg.evaluate_escalation.assert_not_called() rec.assert_not_called() + sup.assert_not_called()