Merge pull request #905 from cbcoutinho/feat/escalation-suppressed-metric

feat(ingest): record suppressed OCR escalations (what-if-OCR signal)
This commit is contained in:
Chris Coutinho
2026-06-13 17:53:07 +02:00
committed by GitHub
6 changed files with 286 additions and 37 deletions
@@ -21,11 +21,37 @@ mapping lives in the queue layer, which imports :class:`EscalateError` from here
from __future__ import annotations
from dataclasses import dataclass
from typing import Literal
# 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: Literal["hop", "suppressed"]
to_tier: str
reason: Literal["empty_text", "low_confidence"]
def next_tier(current: str) -> str | None:
"""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 .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__)
@@ -247,6 +247,13 @@ class ProcessorRegistry:
result, content, settings, record=True, filename=filename
)
# NOTE: the suppressed-escalation metric (document_escalation_suppressed_total,
# the "what-if OCR" signal; Deck #324) is intentionally NOT emitted on this
# inline/memory path -- it is instrumented only on the per-tier external
# path (vector/processor._parse_pdf_tier via evaluate_escalation). When OCR
# is off here the would-be escalation is simply not taken (the gate below);
# operators reading the suppressed counter are on the procrastinate fleet.
#
# 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
@@ -390,22 +397,39 @@ 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_ocr_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_ocr_enabled`` drops only the OCR-enabled gate (not the registered-
processor requirement): it answers "would this tier run if OCR were turned
on?" — used to compute the *ideal* escalation target for the what-if-OCR
suppressed-escalation signal. (Today only ``ocr`` has an enabled gate; a
future per-tier gate would extend the condition below.)
"""
if self._pdf_processor_for_tier(tier) is None:
return False
if tier == "ocr" and not settings.document_ocr_enabled:
if (
not ignore_ocr_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_ocr_enabled: bool = False,
) -> str | None:
"""First escalation target above ``current_tier`` that can actually run.
@@ -413,6 +437,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_ocr_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 +451,9 @@ 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_ocr_enabled=ignore_ocr_enabled
):
return tier
return None
@@ -475,25 +503,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 +544,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_ocr_enabled=True
)
if ideal is not None:
return EscalationDecision("suppressed", ideal, reason)
return None
async def _run_processor(
self,
@@ -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.
+29 -10
View File
@@ -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,
@@ -157,16 +158,34 @@ async def _parse_pdf_tier(
result, content, tier, settings, filename=filename
)
if decision is not None:
to_tier, reason = decision
record_document_escalation(tier, to_tier, reason)
logger.info(
"Escalating %s %s->%s (reason=%s)",
filename or "<bytes>",
tier,
to_tier,
reason,
)
raise EscalateError(from_tier=tier, to_tier=to_tier, reason=reason)
if 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 disabled (reason=%s), "
"indexing at current tier",
filename or "<bytes>",
tier,
decision.to_tier,
decision.reason,
)
else: # "hop" — the Literal kind makes this branch exhaustive.
record_document_escalation(tier, decision.to_tier, decision.reason)
logger.info(
"Escalating %s %s->%s (reason=%s)",
filename or "<bytes>",
tier,
decision.to_tier,
decision.reason,
)
raise EscalateError(
from_tier=tier, to_tier=decision.to_tier, reason=decision.reason
)
return result
+105 -3
View File
@@ -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,105 @@ 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")
def test_evaluate_escalation_terminal_when_ocr_unregistered_and_off(monkeypatch):
"""No OCR processor registered at all (not merely disabled) → genuinely
terminal: returns None, NOT a suppressed decision. 'Absent' != 'disabled'."""
monkeypatch.setattr(reg_mod, "record_document_classification", MagicMock())
r = _registry((_Fake("fast", "fast"), 20)) # only fast; no ocr processor
res = ProcessingResult(
text="",
metadata={
"page_count": 1,
"page_boundaries": [{"page": 1, "start_offset": 0, "end_offset": 0}],
},
processor="fast",
)
assert r.evaluate_escalation(res, b"%PDF", "fast", _Settings(ocr=False)) is None
def test_evaluate_escalation_empty_suppressed_even_when_structured_registered(
monkeypatch,
):
"""empty_text uses minimum='ocr', so it skips structured even when structured
IS registered: with OCR off it suppresses to ocr, never hops to structured
(a text extractor can't conjure text from a raster scan)."""
monkeypatch.setattr(reg_mod, "record_document_classification", MagicMock())
r = _registry(
(_Fake("fast", "fast"), 20),
(_Fake("structured", "structured"), 10), # registered but skipped for empty
(_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")
+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
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()