Classifying OCR Extraction Errors for Manual Review

Once the categorization gate has typed an extraction failure, the record still has to reach a human — and the recurring production bug is that every flagged record lands in one undifferentiated “review” bucket, so a two-minute treasury FX check sits behind hundreds of low-confidence re-keys and reviewers approve blind. This page covers the manual-review triage layer: the deterministic router that turns a categorized failure into a queue-specific review ticket carrying priority, SLA, and just enough context to decide in one look. It is the human-triage detail delegated by the Receipt Error Categorization gate, which is itself the control point inside the broader Receipt Ingestion & OCR Data Extraction framework. Categorization decides what kind of failure a record is; this layer decides who reviews it, how urgently, and with what evidence — without ever mutating the original payload.

Why Standard Approaches Fail

Three named failure modes account for almost every stalled or mis-routed review queue in production accounts-payable and corporate-travel pipelines:

Single-queue dumping. A naive pipeline routes every non-clean record to a shared “needs review” inbox. Because the queue mixes a CURRENCY_CONVERSION_DRIFT that only treasury can adjudicate with a flood of OCR_CONFIDENCE_BELOW_THRESHOLD re-keys that a data clerk clears, high-severity records inherit the SLA of the slowest item in the bucket and freeze reimbursements past close.
Context starvation. The review ticket carries only a record_id. The reviewer has no source_hash to pull the original image, no pointer to the specific field that tripped, and no violation set — so they either re-derive the failure from scratch or rubber-stamp it. Context starvation is what turns a manual-review step into a compliance liability rather than a control.
Severity flattening on collision. A record that is simultaneously low-confidence and missing a mandatory field gets routed by the first rule the loop happens to see, not by severity. The same receipt then appears in two queues, or in the softer one, fracturing the audit trail — the same category-collision hazard the categorization gate resolves upstream, re-introduced at the routing boundary.

The design goal is a router that is deterministic (same categorized record, same queue, same ticket key), that assigns exactly one owning queue by severity while preserving the full violation set as evidence, and that attaches a self-sufficient context bundle so the reviewer never has to reconstruct the failure.

Architecture & Algorithm

The triage layer consumes the CategorizedRecord emitted by the categorization gate — its category, severity, ordered violations set, decision_hash, and source_hash — and maps it to exactly one ReviewTicket. Routing is table-driven so the queue contract is auditable and version-pinned; the ticket key is derived from the upstream decision_hash, which makes ticket creation idempotent (re-processing the same batch never double-queues a receipt). The module below is self-contained and runnable, uses Pydantic v2 for boundary validation, streams through a generator to keep memory flat across a full close-cycle batch, and emits structured JSON audit metadata on every ticket.

from __future__ import annotations

import hashlib
import json
import logging
from datetime import datetime, timedelta, timezone
from enum import Enum
from itertools import islice
from typing import Iterator, Optional

from pydantic import BaseModel, ConfigDict, Field

TRIAGE_VERSION = "manual-review-triage/2026.07"

# --- Structured audit logging -------------------------------------------------

class _TicketJSONFormatter(logging.Formatter):
    """One machine-readable JSON object per triage decision so a SIEM or
    compliance warehouse ingests it without regex parsing."""

    def format(self, record: logging.LogRecord) -> str:
        payload = {
            "ts": datetime.now(timezone.utc).isoformat(),
            "level": record.levelname,
            "stage": "manual_review_triage",
            "record_id": getattr(record, "record_id", "UNKNOWN"),
            "review_queue": getattr(record, "review_queue", "UNKNOWN"),
            "priority": getattr(record, "priority", "P4"),
            "ticket_key": getattr(record, "ticket_key", ""),
            "decision_hash": getattr(record, "decision_hash", ""),
            "triage_version": getattr(record, "triage_version", ""),
            "msg": record.getMessage(),
        }
        return json.dumps(payload, separators=(",", ":"))


logger = logging.getLogger("expense.manual_review_triage")
if not logger.handlers:
    _handler = logging.StreamHandler()
    _handler.setFormatter(_TicketJSONFormatter())
    logger.addHandler(_handler)
    logger.setLevel(logging.INFO)


# --- Inputs and taxonomy ------------------------------------------------------

class ReviewQueue(str, Enum):
    MANUAL_VISION_REKEY = "manual_vision_rekey"   # verify against raw image
    AP_EXCEPTION = "ap_exception"                 # supply a missing field
    TREASURY_FX = "treasury_fx"                   # adjudicate currency drift
    TRAVEL_POLICY_AUDIT = "travel_policy_audit"   # reconcile line items
    COMPLIANCE_HOLD = "compliance_hold"           # hard-stop, frozen payout


class CategorizedRecord(BaseModel):
    """The upstream contract emitted by the categorization gate. Extra keys are
    rejected so a contract change surfaces here instead of propagating."""

    model_config = ConfigDict(extra="forbid")

    record_id: str
    category: str
    severity: str                       # CRITICAL | HIGH | MEDIUM | LOW | INFO
    violations: list[str]               # full tripped-rule set, for evidence
    decision_hash: str
    source_hash: str = ""               # SHA-256 of the original artifact
    failed_fields: list[str] = Field(default_factory=list)


class ReviewTicket(BaseModel):
    record_id: str
    review_queue: ReviewQueue
    priority: str                       # P1 (fastest) .. P4
    sla_deadline_utc: str
    review_reason: str
    evidence: dict                      # self-sufficient context bundle
    decision_hash: str
    ticket_key: str
    triage_version: str
    created_at_utc: str


# --- Routing contract ---------------------------------------------------------
# category -> (queue, priority, SLA hours, human-readable review reason).
# Severity is carried on the record; priority/SLA are the queue's operating
# contract and are what the reviewer's dashboard sorts on.
_ROUTES: dict[str, tuple[ReviewQueue, str, int, str]] = {
    "POLICY_PRECHECK_VIOLATION":     (ReviewQueue.COMPLIANCE_HOLD,     "P1", 2,  "Confirm the hard-stop breach before any release."),
    "SCHEMA_VALIDATION_FAILURE":     (ReviewQueue.COMPLIANCE_HOLD,     "P1", 2,  "Upstream contract broke; do not release."),
    "MISSING_MANDATORY_FIELD":       (ReviewQueue.AP_EXCEPTION,        "P2", 8,  "Supply the missing field from the source image."),
    "CURRENCY_CONVERSION_DRIFT":     (ReviewQueue.TREASURY_FX,         "P2", 8,  "Re-derive the converted amount at the booking rate."),
    "LINE_ITEM_ARITHMETIC_MISMATCH": (ReviewQueue.TRAVEL_POLICY_AUDIT, "P3", 24, "Reconcile line items and tax against the total."),
    "OCR_CONFIDENCE_BELOW_THRESHOLD":(ReviewQueue.MANUAL_VISION_REKEY, "P4", 24, "Verify low-confidence fields against the raw image."),
}

# Fallback for any category not in the contract: fail closed, not open.
_FALLBACK: tuple[ReviewQueue, str, int, str] = (
    ReviewQueue.COMPLIANCE_HOLD, "P1", 2, "Unmapped category; route to compliance and freeze.",
)


class ManualReviewTriage:
    """Map one CategorizedRecord to exactly one ReviewTicket, deterministically."""

    def __init__(self, *, triage_version: str = TRIAGE_VERSION) -> None:
        self._triage_version = triage_version

    def _ticket_key(self, decision_hash: str) -> str:
        # Derived from the upstream decision hash so re-processing the same
        # record produces the same ticket key -> idempotent, never double-queued.
        seed = f"{decision_hash}|{self._triage_version}"
        return hashlib.sha256(seed.encode("utf-8")).hexdigest()[:32]

    def triage(self, record: CategorizedRecord) -> ReviewTicket:
        queue, priority, sla_hours, reason = _ROUTES.get(record.category, _FALLBACK)
        now = datetime.now(timezone.utc)
        # Evidence bundle: everything the reviewer needs without a round trip.
        evidence = {
            "source_hash": record.source_hash,      # resolves to the raw image
            "failed_fields": record.failed_fields,  # what to look at first
            "violations": record.violations,        # full audit context
            "severity": record.severity,
            "category": record.category,
        }
        ticket = ReviewTicket(
            record_id=record.record_id,
            review_queue=queue,
            priority=priority,
            sla_deadline_utc=(now + timedelta(hours=sla_hours)).isoformat(),
            review_reason=reason,
            evidence=evidence,
            decision_hash=record.decision_hash,
            ticket_key=self._ticket_key(record.decision_hash),
            triage_version=self._triage_version,
            created_at_utc=now.isoformat(),
        )
        logger.info(
            "review_ticket_created",
            extra={
                "record_id": ticket.record_id,
                "review_queue": ticket.review_queue.value,
                "priority": ticket.priority,
                "ticket_key": ticket.ticket_key,
                "decision_hash": ticket.decision_hash,
                "triage_version": ticket.triage_version,
            },
        )
        return ticket

    def triage_stream(
        self, records: Iterator[dict], *, chunk_size: int = 500
    ) -> Iterator[ReviewTicket]:
        """Yield tickets in bounded chunks so memory stays flat at close-cycle
        scale; CLEAN_PASS / INFO records are skipped — they never need a human."""
        while chunk := list(islice(records, chunk_size)):
            for raw in chunk:
                record = CategorizedRecord.model_validate(raw)
                if record.severity == "INFO":
                    continue
                yield self.triage(record)


if __name__ == "__main__":
    triage = ManualReviewTriage()
    flagged = [
        {"record_id": "R2", "category": "OCR_CONFIDENCE_BELOW_THRESHOLD",
         "severity": "LOW", "violations": ["OCR_CONFIDENCE_BELOW_THRESHOLD"],
         "decision_hash": "d2", "source_hash": "def", "failed_fields": ["total_amount"]},
        {"record_id": "R3", "category": "MISSING_MANDATORY_FIELD",
         "severity": "HIGH", "violations": ["MISSING_MANDATORY_FIELD",
         "OCR_CONFIDENCE_BELOW_THRESHOLD"], "decision_hash": "d3",
         "source_hash": "ghi", "failed_fields": ["merchant"]},
        {"record_id": "R4", "category": "POLICY_PRECHECK_VIOLATION",
         "severity": "CRITICAL", "violations": ["POLICY_PRECHECK_VIOLATION"],
         "decision_hash": "d4", "source_hash": "jkl", "failed_fields": []},
    ]
    for t in triage.triage_stream(iter(flagged)):
        print(t.record_id, t.review_queue.value, t.priority, t.sla_deadline_utc)

Because the terminal category is already the highest-severity violation (the categorization gate resolves collisions before this layer sees the record), routing keys off a single field and cannot flatten severity: R3 carries both a missing field and low confidence in its violations, but its category is MISSING_MANDATORY_FIELD, so it routes once — to the AP exception queue at P2 — while the low-confidence finding rides along as evidence rather than spawning a second ticket. Folding decision_hash into ticket_key makes ticket creation idempotent, so a replayed batch reuses the existing ticket instead of duplicating it.

Step-by-Step Integration

Consume the categorized stream, not raw OCR. Wire this layer strictly after the Receipt Error Categorization gate so every record already carries a terminal category, severity, violations, and decision_hash. Never re-derive the failure here.
Pin the routing contract. Store _ROUTES and TRIAGE_VERSION in your config store and bump the version whenever a queue mapping or SLA changes, so audit reconstruction resolves to one routing table.
Skip clean records. Confirm INFO/CLEAN_PASS records are filtered before triage — they advance to the Core Policy Architecture & Taxonomy Design engine, not a human queue.
Attach the evidence bundle to the ticket, not a pointer to it. Resolve source_hash to a pre-signed image URL and inline failed_fields + violations so the reviewer decides without a second round trip.

Enforce idempotency at the ticketing boundary. Upsert on ticket_key; a re-run of the batch must update the existing ticket rather than create a duplicate. Verify with:

from manual_review_triage import CategorizedRecord, ManualReviewTriage
raw = {"record_id": "R", "category": "MISSING_MANDATORY_FIELD",
       "severity": "HIGH", "violations": ["MISSING_MANDATORY_FIELD"],
       "decision_hash": "d1", "source_hash": "h", "failed_fields": ["merchant"]}
t = ManualReviewTriage()
rec = CategorizedRecord.model_validate(raw)
assert t.triage(rec).ticket_key == t.triage(rec).ticket_key   # idempotent
assert t.triage(rec).review_queue.value == "ap_exception"      # right queue

Mirror the ticket to the audit ledger. Append every review_ticket_created JSON event to the same append-only ledger the categorization gate writes to, so the decision_hash links the auto-decision and the human decision into one chain of custody.
Preserve reviewer overrides as new records. When a reviewer resolves a ticket, append the resolution, reviewer_id, and resolved_at as a separate compliance record keyed by record_id — never mutate the original OCR payload or the categorized record.

Edge Cases & Gotchas

Edge condition	What breaks	Mitigation
Timezone-shifted `transaction_date`	A UTC-normalized date crosses a submission-window boundary and re-flags on replay, spawning a second ticket	Normalize to UTC upstream; key the ticket on `decision_hash`, never on wall-clock time
Split-tender receipts	Line items reconcile to a partial total, so the same expense produces two low-total tickets in different queues	Group by `source_hash` before triage; route the group to a single travel-policy audit ticket
Duplicate flagged across windows	The same receipt resubmitted in two periods creates two review tickets	Cross-check against the Duplicate Receipt Detection pipeline before opening a ticket
FX variance near tolerance	A `CURRENCY_CONVERSION_DRIFT` oscillates in and out of tolerance on re-processing, churning treasury tickets	Snapshot the reference rate on the record; align the drift band with Dynamic Threshold Tuning
Faded thermal / very low confidence	Re-keying stalls because no field is legible enough to verify	Route to `MANUAL_VISION_REKEY` with a resubmission fallback rather than looping the OCR retry
Unmapped `category`	A new taxonomy code has no route and could silently drop	Fail closed to `COMPLIANCE_HOLD` (the `_FALLBACK`), never to a soft queue
MCC ambiguity on precheck	A borderline merchant category over-triggers `COMPLIANCE_HOLD`	Resolve the code first via Merchant Category Code routing

FAQ

How is this different from the receipt error categorization gate?

Categorization is the deterministic classification step: it types every extraction result into one error category with a severity and a decision hash. This layer is the routing step that runs only on the flagged subset — it takes that category and decides which human queue owns the record, at what priority and SLA, and packages the evidence a reviewer needs. Classification answers “what is wrong”; triage answers “who fixes it and how urgently”.

Why route by category instead of re-evaluating the rules here?

Re-evaluating rules at the routing boundary re-introduces severity flattening: two components can disagree on which violation wins, and the record fractures across queues. Because the Receipt Error Categorization gate already resolves collisions to a single highest-severity terminal category, keying off that one field guarantees one owning queue and one ticket per record.

What stops the same receipt from creating duplicate review tickets?

The ticket_key is a SHA-256 of the upstream decision_hash plus the triage version, and the ticketing system upserts on it. Since categorization is idempotent — the same artifact under the same policy version yields the same decision_hash — a replayed batch reuses the existing ticket instead of opening a new one.

How should reviewer decisions be recorded for audit?

Append the resolution as a separate compliance record keyed by record_id, capturing reviewer_id, resolution, and resolved_at. Never mutate the original OCR payload or the categorized record; the decision_hash links the automated decision and the human decision into one immutable chain of custody suitable for Sarbanes-Oxley Act control testing.

Where do low-confidence records go when re-keying can’t recover the value?

They start in the MANUAL_VISION_REKEY queue against the raw image. If the field is illegible (faded thermal ink, cropped capture), the reviewer triggers a resubmission fallback rather than looping the OCR retry — align the confidence floor that produces these with your Tesseract OCR configuration so the queue volume stays predictable.

Receipt Error Categorization — the parent gate that types failures before this layer routes them
Tesseract OCR configuration — sets the confidence floor that governs manual-review volume
pdfplumber line-item parsing — produces the line-item structure behind arithmetic-mismatch tickets
Async batch processing — the chunked-streaming discipline this router reuses
Duplicate Receipt Detection — cross-check before opening a ticket for a resubmitted receipt