feat(ingest): record suppressed OCR escalations (what-if-OCR signal)

OCR is the paid, opt-in tier (DOCUMENT_OCR_ENABLED, default off). The per-tier
escalation gate already declines to hop to OCR when it's disabled (the pre-OCR
tier is terminal — no surprise cost), but that left operators blind to how much
OCR demand exists.

evaluate_escalation now returns a structured EscalationDecision:
- "hop"        — a higher tier can run; the caller raises EscalateError (queue-hop).
- "suppressed" — the ideal next tier (e.g. ocr) exists but is DISABLED; the caller
                 indexes the current tier's output as terminal and records the
                 would-be hop on the new astrolabe_document_escalation_suppressed_total
                 {from_tier,to_tier,reason} counter instead of hopping.
- None         — index as-is (good text, or no such tier at all).

So with OCR off, 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. next_available_tier gains an
ignore_enabled flag to compute the *ideal* (enabled-gate-ignored) target.

Tests: registry suppressed vs hop vs terminal (incl. structured-hop-not-suppressed
when OCR off but structured available); _parse_pdf_tier records suppressed +
indexes without raising.

Deck #324 (parent #323).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Chris Coutinho
2026-06-13 15:16:03 +02:00
co-authored by Claude Opus 4.8
parent 6db48830af
commit a27ddb2d5a
6 changed files with 227 additions and 33 deletions
@@ -21,11 +21,36 @@ mapping lives in the queue layer, which imports :class:`EscalateError` from here
from __future__ import annotations from __future__ import annotations
from dataclasses import dataclass
# Cheapest-first. ``llm`` is reserved (see base.DocumentProcessor.tier) and not # Cheapest-first. ``llm`` is reserved (see base.DocumentProcessor.tier) and not
# wired yet, so it is intentionally absent from the live ladder. # wired yet, so it is intentionally absent from the live ladder.
TIER_LADDER: tuple[str, ...] = ("fast", "structured", "ocr") 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: def next_tier(current: str) -> str | None:
"""The next tier above ``current`` in the ladder, or ``None`` if terminal. """The next tier above ``current`` in the ladder, or ``None`` if terminal.
@@ -15,7 +15,7 @@ from nextcloud_mcp_server.observability.tracing import trace_operation
from .base import DocumentProcessor, ProcessingResult, ProcessorError from .base import DocumentProcessor, ProcessingResult, ProcessorError
from .classifier import DocClassification, classify_from_text, image_coverage_per_page 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__) logger = logging.getLogger(__name__)
@@ -390,22 +390,34 @@ class ProcessorRegistry:
) )
return classification return classification
def _tier_available(self, tier: str, settings: Any) -> bool: def _tier_available(
"""Whether ``tier`` can actually run a PDF parse right now. 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 A tier is available when it has a registered PDF processor and is
enabled; the ``ocr`` tier additionally requires ``DOCUMENT_OCR_ENABLED`` enabled; the ``ocr`` tier additionally requires ``DOCUMENT_OCR_ENABLED``
(so OCR stays opt-in and a misconfigured tenant never escalates to a (so OCR stays opt-in and a misconfigured tenant never escalates to a
backend it hasn't turned on). 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: if self._pdf_processor_for_tier(tier) is None:
return False 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 False
return True return True
def next_available_tier( 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: ) -> str | None:
"""First escalation target above ``current_tier`` that can actually run. """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 ``minimum``'s rung, when given) and returns the first
:meth:`_tier_available` tier. ``None`` means no higher tier can run -- :meth:`_tier_available` tier. ``None`` means no higher tier can run --
``current_tier`` is then terminal and its result is indexed as-is. ``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: try:
cur_idx = TIER_LADDER.index(current_tier) cur_idx = TIER_LADDER.index(current_tier)
@@ -425,7 +439,7 @@ class ProcessorRegistry:
except ValueError: except ValueError:
pass pass
for tier in TIER_LADDER[start_idx:]: 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 tier
return None return None
@@ -475,25 +489,33 @@ class ProcessorRegistry:
settings: Any, settings: Any,
*, *,
filename: str | None = None, filename: str | None = None,
) -> tuple[str, str] | None: ) -> EscalationDecision | None:
"""Decide whether ``current_tier``'s result must escalate (external path). """Decide whether ``current_tier``'s result must escalate (external path).
Returns ``(to_tier, reason)`` when the parse is too poor to index and a Returns an :class:`EscalationDecision` (``"hop"`` or ``"suppressed"``)
higher tier can run, else ``None`` (index the result as-is). Reuses the when the classifier judges the parse too poor to index, else ``None``
tier-0 classifier as the post-parse quality gate, so the escalation (index as-is). Reuses the tier-0 classifier as the post-parse quality
signal is identical to the inline pipeline's. gate, so the signal is identical to the inline pipeline's.
A hard parse FAILURE (``result.success`` False) is never escalated: a A hard parse FAILURE (``result.success`` False) is never escalated: a
corrupt/encrypted PDF one engine can't open usually defeats the others 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. 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`` - ``total_chars == 0`` (scanned / no text layer) -> target the ``ocr``
tier directly. Text-extractor tiers (``structured``) cannot conjure tier directly. Text-extractor tiers (``structured``) cannot conjure
text from a pure raster scan, so a structured hop would just be wasted. 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 - low-confidence but non-empty layer -> escalate to the next rung, so a
different in-cluster extractor can try before paying for OCR. 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( classification = self._classify_result(
result, result,
@@ -508,14 +530,22 @@ class ProcessorRegistry:
if classification.page_count <= 0: if classification.page_count <= 0:
return None return None
if classification.total_chars == 0: if classification.total_chars == 0:
to_tier = self.next_available_tier(current_tier, settings, minimum="ocr") minimum: str | None = "ocr"
reason = "empty_text" reason = "empty_text"
else: else:
to_tier = self.next_available_tier(current_tier, settings) minimum = None
reason = "low_confidence" reason = "low_confidence"
if to_tier is None: 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 return None
return (to_tier, reason)
async def _run_processor( async def _run_processor(
self, self,
@@ -276,6 +276,21 @@ document_escalation_total = Counter(
["from_tier", "to_tier", "reason"], ["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 # 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 # timeout/OOM that kills the worker is caught here. This is distinct from
# ``document_parse_total{status="error"}`` (an in-process exception): a hard # ``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() ).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: def record_document_parse_failed(reason: str) -> None:
"""Record a hard parse failure from the isolated worker. """Record a hard parse failure from the isolated worker.
+25 -6
View File
@@ -28,6 +28,7 @@ from nextcloud_mcp_server.models.deck import DeckCard
from nextcloud_mcp_server.observability.metrics import ( from nextcloud_mcp_server.observability.metrics import (
record_document_chunks, record_document_chunks,
record_document_escalation, record_document_escalation,
record_document_escalation_suppressed,
record_document_parse_failed, record_document_parse_failed,
record_embedding, record_embedding,
record_embedding_tokens, record_embedding_tokens,
@@ -156,17 +157,35 @@ async def _parse_pdf_tier(
decision = registry.evaluate_escalation( decision = registry.evaluate_escalation(
result, content, tier, settings, filename=filename result, content, tier, settings, filename=filename
) )
if decision is not None: if decision is not None and decision.kind == "suppressed":
to_tier, reason = decision # The ideal next tier (e.g. ocr) is disabled, so we do NOT hop: index
record_document_escalation(tier, to_tier, reason) # 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 "<bytes>",
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( logger.info(
"Escalating %s %s->%s (reason=%s)", "Escalating %s %s->%s (reason=%s)",
filename or "<bytes>", filename or "<bytes>",
tier, tier,
to_tier, decision.to_tier,
reason, 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 return result
+65 -3
View File
@@ -13,6 +13,7 @@ from nextcloud_mcp_server.document_processors.base import (
DocumentProcessor, DocumentProcessor,
ProcessingResult, ProcessingResult,
) )
from nextcloud_mcp_server.document_processors.escalation import EscalationDecision
from nextcloud_mcp_server.document_processors.registry import ProcessorRegistry from nextcloud_mcp_server.document_processors.registry import ProcessorRegistry
pytestmark = pytest.mark.unit pytestmark = pytest.mark.unit
@@ -334,7 +335,7 @@ def test_evaluate_escalation_empty_jumps_to_ocr(monkeypatch):
processor="fast", processor="fast",
) )
decision = r.evaluate_escalation(res, b"%PDF", "fast", _Settings(ocr=True)) 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): def test_evaluate_escalation_lowconf_goes_to_structured(monkeypatch):
@@ -357,7 +358,7 @@ def test_evaluate_escalation_lowconf_goes_to_structured(monkeypatch):
processor="fast", processor="fast",
) )
decision = r.evaluate_escalation(res, b"%PDF", "fast", _Settings(ocr=True)) 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): 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", processor="fast",
) )
decision = r.evaluate_escalation(res, b"%PDF", "fast", _Settings(ocr=True)) 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")
+36 -7
View File
@@ -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 ``processor._parse_pdf_tier`` runs one tier and either indexes the result,
index or raises ``EscalateError`` (a queue-hop). These exercise the decision raises ``EscalateError`` (a real queue-hop), or — when the ideal next tier is
without standing up the full ingest pipeline. 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 from unittest.mock import AsyncMock, MagicMock
@@ -10,7 +11,10 @@ from unittest.mock import AsyncMock, MagicMock
import pytest import pytest
from nextcloud_mcp_server.document_processors.base import ProcessingResult 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 from nextcloud_mcp_server.vector import processor
pytestmark = pytest.mark.unit pytestmark = pytest.mark.unit
@@ -25,7 +29,9 @@ def _registry(result: ProcessingResult, decision):
async def test_good_parse_returns_result(monkeypatch): async def test_good_parse_returns_result(monkeypatch):
rec = MagicMock() rec = MagicMock()
sup = MagicMock()
monkeypatch.setattr(processor, "record_document_escalation", rec) monkeypatch.setattr(processor, "record_document_escalation", rec)
monkeypatch.setattr(processor, "record_document_escalation_suppressed", sup)
result = ProcessingResult(text="clean", metadata={}, processor="fast") result = ProcessingResult(text="clean", metadata={}, processor="fast")
reg = _registry(result, decision=None) reg = _registry(result, decision=None)
out = await processor._parse_pdf_tier( out = await processor._parse_pdf_tier(
@@ -33,13 +39,14 @@ async def test_good_parse_returns_result(monkeypatch):
) )
assert out is result assert out is result
rec.assert_not_called() rec.assert_not_called()
sup.assert_not_called()
async def test_low_quality_parse_raises_escalate(monkeypatch): async def test_low_quality_parse_raises_escalate(monkeypatch):
rec = MagicMock() rec = MagicMock()
monkeypatch.setattr(processor, "record_document_escalation", rec) monkeypatch.setattr(processor, "record_document_escalation", rec)
result = ProcessingResult(text="", metadata={}, processor="fast") 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: with pytest.raises(EscalateError) as ei:
await processor._parse_pdf_tier( await processor._parse_pdf_tier(
reg, b"%PDF", "application/pdf", "f.pdf", "fast", settings=object() 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") 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): async def test_hard_failure_returns_result_without_escalating(monkeypatch):
rec = MagicMock() rec = MagicMock()
sup = MagicMock()
monkeypatch.setattr(processor, "record_document_escalation", rec) monkeypatch.setattr(processor, "record_document_escalation", rec)
monkeypatch.setattr(processor, "record_document_escalation_suppressed", sup)
result = ProcessingResult( result = ProcessingResult(
text="", text="",
metadata={"parse_failed_reason": "oversize"}, metadata={"parse_failed_reason": "oversize"},
processor="size_guard", processor="size_guard",
success=False, success=False,
) )
reg = _registry(result, decision=("ocr", "empty_text")) reg = _registry(result, decision=EscalationDecision("hop", "ocr", "empty_text"))
out = await processor._parse_pdf_tier( out = await processor._parse_pdf_tier(
reg, b"%PDF", "application/pdf", "big.pdf", "fast", settings=object() 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 assert out is result
reg.evaluate_escalation.assert_not_called() reg.evaluate_escalation.assert_not_called()
rec.assert_not_called() rec.assert_not_called()
sup.assert_not_called()