Error Handling Protocols

Construction change-order automation sits at the intersection of financial liability, schedule compression, and regulatory compliance. When automated document ingestion meets a malformed PDF, a corrupted Excel takeoff, or a misaligned extracted field, the damage does not stay local: it cascades into incorrect cost tracking, stalled approvals, and audit exposure that surfaces months later during a payment dispute. The specific sub-problem this page solves is how a parsing pipeline should fail — how it detects bad data at the boundary, recovers from partial extraction without inventing numbers, and escalates the residue to a human without losing the document. A resilient protocol enforces strict schema boundaries before parsing, runs deterministic fallbacks during extraction, validates arithmetic against contractual tolerances, and routes every exception through role-aware channels with bounded retries. This page is the error-handling subsystem of the Automated Document Ingestion & Parsing pipeline, and it assumes the upstream gateway has already classified each document and bound it to a project taxonomy.

Prerequisites

This subsystem builds on the rest of the ingestion stack. Before implementing the patterns below, you need:

• Python 3.11+ with pydantic v2 for typed validation, tenacity for declarative retry control, and the standard decimal, logging, and enum modules.
• A task queue — Celery with a Redis or RabbitMQ broker — so that failed work can be retried out of band and parked in a dead-letter queue rather than blocking a request thread. This is the same broker described in async batching workflows.
• Parsing dependencies already wired up: pdfplumber / PyMuPDF for native PDF text, pytesseract for the OCR preprocessing fallback, and the coordinate-mapping logic from field extraction techniques.
• An established schema contract. Error handling is only as good as the contract it enforces; the canonical change-order model lives in schema validation rules and is imported here rather than redefined.
• A confidence score on every extracted field. Routing depends on it. The pipeline uses three site-canonical bands everywhere: 0.92 and above auto-routes, 0.75–0.92 goes to human review, and below 0.75 is quarantined.

The protocol also assumes documents have already been mapped to a work breakdown structure code at the gateway, so an exception payload can name the cost account it would have touched.

Architecture: inputs, stages, and error branches

Error handling is not a single try/except — it is four ordered stages, each with its own failure branch. A document that clears one stage can still fail the next, and the protocol’s job is to make sure every failure mode produces a structured, replayable exception rather than a silent default. The diagram below traces a change-order payload from raw bytes to either a committed ledger entry or a parked exception.

The four error branches map to four error-code families, summarized here:

Step-by-step implementation

Step 1 — Gatekeep at ingestion with a strict schema

The first checkpoint rejects structurally unsound documents before they reach the calculation engine. Change orders arrive as scanned PDFs with overlapping wet signatures, stamped revision blocks, and non-standard headers; rather than letting a malformed payload propagate, the ingestion layer enforces a Pydantic contract. Mandatory fields such as change_order_number, original_contract_value, revised_total, approval_status, and line_items get explicit type coercion and nullability rules. When validation fails, the system captures the exact violation path, logs the raw payload hash for auditability, and halts. The middleware returns structured error codes — SCHEMA_MISSING_REQUIRED_FIELD, SCHEMA_TYPE_MISMATCH — instead of generic exceptions, so downstream routing can distinguish a recoverable formatting issue from fatal corruption. Following Python’s exception handling guidelines keeps the custom error hierarchy interoperable with standard logging and enterprise SIEM integrations.

import hashlib
import logging
from decimal import Decimal, InvalidOperation
from typing import Any, Literal
from pydantic import BaseModel, Field, ValidationError, field_validator

logger = logging.getLogger("error_handling")

# Discipline codes are site-canonical and constrained, never free strings.
Discipline = Literal["ARCH", "STR", "MEP", "CIV", "ELEC", "PLMB"]

class LineItem(BaseModel):
    description: str = Field(..., min_length=1)
    discipline: Discipline
    quantity: Decimal = Field(..., ge=0)
    unit_price: Decimal = Field(..., ge=0)
    total: Decimal = Field(..., ge=0)

class ChangeOrderSchema(BaseModel):
    change_order_number: str = Field(..., min_length=3, max_length=20)
    # WBS element pattern: PROJ-NNN-DIV-NN (project, sequence, MasterFormat division)
    wbs_code: str = Field(..., pattern=r"^[A-Z]{2,5}-\d{3}-\d{2}-\d{2}$")
    original_contract_value: Decimal = Field(..., ge=0)
    revised_total: Decimal = Field(..., ge=0)
    approval_status: Literal["PENDING", "UNDER_REVIEW", "APPROVED", "REJECTED", "EXECUTED"]
    line_items: list[LineItem]

    @field_validator("revised_total")
    @classmethod
    def validate_calculation(cls, v: Decimal, info: Any) -> Decimal:
        # Cross-field check: revised total must equal the line-item sum
        # within a contractual tolerance (see Step 3).
        if "line_items" in info.data:
            calculated_total = sum(item.total for item in info.data["line_items"])
            tolerance = Decimal("0.05")
            if abs(v - calculated_total) > tolerance:
                raise ValueError(
                    f"Revised total {v} deviates from calculated sum {calculated_total} "
                    f"beyond ±{tolerance} tolerance."
                )
        return v

def validate_ingestion_payload(raw_payload: dict[str, Any]) -> dict[str, Any]:
    """Validate a raw change-order payload against strict schema boundaries.

    Returns a structured status dict for downstream routing rather than
    raising, so the exception router can branch on the error code.
    """
    # Hash is computed BEFORE the try block so it is available on every path.
    payload_hash = hashlib.sha256(str(raw_payload).encode("utf-8")).hexdigest()[:12]
    try:
        validated_co = ChangeOrderSchema(**raw_payload)
        return {
            "status": "VALIDATED",
            "co_number": validated_co.change_order_number,
            "payload_hash": payload_hash,
            "next_stage": "PERSISTENCE",
        }
    except ValidationError as exc:
        error_loc = exc.errors()[0]["loc"]
        error_msg = exc.errors()[0]["msg"]
        logger.error("Schema violation at %s: %s | hash=%s", error_loc, error_msg, payload_hash)
        return {
            "status": "REJECTED",
            "error_code": "SCHEMA_VALIDATION_FAILURE",
            "violation_path": error_loc,
            "payload_hash": payload_hash,
            "details": error_msg,
        }
    except InvalidOperation:
        logger.error("Non-numeric financial data during Decimal coercion | hash=%s", payload_hash)
        return {
            "status": "REJECTED",
            "error_code": "SCHEMA_TYPE_MISMATCH",
            "payload_hash": payload_hash,
            "details": "Invalid numeric format in financial fields.",
        }

Note that payload_hash is computed before the try block so it is available on both the success and the error return paths — an exception you cannot trace back to a specific document is useless during an audit.

Step 2 — Recover from extraction failures without inventing data

Once a document is structurally sound, the parsing layer has to survive field-level failures without collapsing the workflow. Construction documents rarely conform to rigid templates: subcontractors submit Excel files with merged cells, hidden rows, and macro-driven formulas that break standard tabular readers. The extraction engine therefore uses a tiered fallback. Primary extraction relies on coordinate-based bounding boxes trained on historical change-order layouts. If the confidence score drops below 0.75 for a critical financial field, the system triggers an OCR preprocessing pass with layout-aware segmentation. When OCR still cannot resolve a field, the parser must not default to zero or null — the cardinal sin of construction automation. Instead it flags the specific cell, preserves the raw image crop, and routes the payload to the human-review queue.

This is where the protocol couples to the rest of the pipeline. Integration with PDF/Excel sync pipelines ensures a fallback state does not desynchronize version control across the two document formats, and the confidence metadata produced by field extraction techniques rides directly on the exception payload so a document-control specialist can triage in seconds.

from dataclasses import dataclass, field

@dataclass
class FieldExtraction:
    name: str
    value: Decimal | str | None
    confidence: float
    raw_crop_ref: str | None = None  # storage pointer to the image region

# Site-canonical routing bands — used identically across every subsystem.
AUTO_ROUTE = 0.92
HUMAN_REVIEW = 0.75

def classify_extraction(fields: list[FieldExtraction]) -> dict[str, Any]:
    """Decide a routing state from the weakest field's confidence."""
    weakest = min(fields, key=lambda f: f.confidence)
    if weakest.confidence >= AUTO_ROUTE:
        return {"state": "AUTO_ROUTE", "next_stage": "CALCULATION_CHECK"}
    if weakest.confidence >= HUMAN_REVIEW:
        return {
            "state": "HUMAN_REVIEW",
            "error_code": "EXTRACTION_LOW_CONFIDENCE",
            "field": weakest.name,
            "confidence": weakest.confidence,
        }
    # Below 0.75: never guess. Preserve the crop and quarantine.
    return {
        "state": "QUARANTINE",
        "error_code": "OCR_UNRESOLVED_FIELD",
        "field": weakest.name,
        "raw_crop_ref": weakest.raw_crop_ref,
    }

Step 3 — Validate arithmetic against contractual tolerances

With fields extracted, the system verifies arithmetic integrity against contractual baselines. Change orders often contain nested formulas, tax multipliers, or retention percentages that deviate from simple addition, so a deterministic routine cross-references the extracted line items against revised_total and applies a configurable tolerance (±0.05 of the document currency, or a percentage for large contracts). A discrepancy beyond tolerance raises a CALCULATION_MISMATCH, which halts approval routing until an estimator reconciles the variance. This is the step that stops silent data corruption from reaching cost-forecasting dashboards — a transposed decimal that survives to the ledger is far more expensive than one caught at ingestion. The cross-field validator embedded in Step 1’s schema is the first line of this check; a standalone routine handles tax and retention math that Pydantic field validators are not the right place for.

Step 4 — Route exceptions deterministically with bounded retries

Exception routing must be deterministic and role-aware. Financial discrepancies route to lead estimators; schema and OCR failures route to document-control specialists; everything else goes to platform engineering. The alerting layer batches through the async queue to prevent notification storms during high-volume submission windows. For transient infrastructure faults — API timeouts, storage rate limits, webhook delivery failures — the system applies exponential backoff with jitter, and persistent failures trip a circuit breaker to keep the queue from saturating. The broader policy for who-gets-paged and when lives in fallback alert routing; the network-level retry mechanics for upstream pulls are detailed in implementing retry logic for failed API document pulls. Leveraging Pydantic’s validation engine alongside structured alerting keeps exception payloads strongly typed and machine-readable across microservice boundaries.

import time
import random
from enum import Enum
from typing import Any, Callable

class AlertChannel(Enum):
    ESTIMATOR = "estimator_team_slack"
    DOC_CONTROL = "doc_control_email"
    DEV_OPS = "platform_engineering_webhook"

@dataclass
class ExceptionRoutingPayload:
    error_code: str
    severity: Literal["INFO", "WARNING", "CRITICAL"]
    document_id: str
    wbs_code: str
    context_metadata: dict[str, Any] = field(default_factory=dict)

def exponential_backoff_with_jitter(max_retries: int = 3, base_delay: float = 1.0) -> Callable:
    """Decorator: exponential backoff + uniform jitter for alert dispatch."""
    def decorator(func: Callable) -> Callable:
        def wrapper(*args: Any, **kwargs: Any) -> Any:
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except Exception as exc:  # transport-level fault, retryable
                    delay = (base_delay * (2 ** attempt)) + random.uniform(0, 0.5)
                    logger.warning(
                        "Alert dispatch %d/%d failed: %s. Retrying in %.2fs",
                        attempt + 1, max_retries, exc, delay,
                    )
                    time.sleep(delay)
            # Retries exhausted: caller trips the circuit breaker + DLQ.
            raise RuntimeError("Max retries exceeded for alert routing pipeline")
        return wrapper
    return decorator

def determine_routing_channel(error_code: str) -> AlertChannel:
    """Map a structured error code to a role-aware notification channel."""
    if error_code.startswith("CALCULATION"):
        return AlertChannel.ESTIMATOR
    if error_code.startswith(("SCHEMA", "OCR", "EXTRACTION")):
        return AlertChannel.DOC_CONTROL
    return AlertChannel.DEV_OPS

@exponential_backoff_with_jitter(max_retries=3, base_delay=1.5)
def dispatch_exception_alert(payload: ExceptionRoutingPayload) -> bool:
    """Route an exception payload to its channel with retry resilience.

    In production, replace the simulated call with a webhook/email SDK.
    """
    target = determine_routing_channel(payload.error_code)
    logger.info("Routing %s [%s] to %s", payload.error_code, payload.document_id, target.value)
    if payload.severity == "CRITICAL":
        raise ConnectionError("Temporary notification gateway timeout")
    return True

Error-code reference

Routing logic keys entirely off the error-code prefix, so the vocabulary is part of the contract. Keep it stable; downstream channel mappings and SIEM rules depend on these exact prefixes.

Confidence-band configuration keys, used identically across every subsystem:

Verification and testing

Confirm each branch behaves deterministically with focused assertions. The point is to prove that bad input produces a structured outcome, never an uncaught exception or a silent zero.

def test_schema_rejection_returns_structured_code():
    bad = {"change_order_number": "X"}  # too short, missing fields
    result = validate_ingestion_payload(bad)
    assert result["status"] == "REJECTED"
    assert result["error_code"].startswith("SCHEMA")
    assert "payload_hash" in result  # always traceable

def test_calculation_tolerance_boundary():
    items = [LineItem(description="Rebar", discipline="STR",
                      quantity=Decimal("10"), unit_price=Decimal("100"),
                      total=Decimal("1000"))]
    # 0.04 drift is within tolerance; 0.06 is not.
    ChangeOrderSchema(change_order_number="CO-2024-089", wbs_code="PROJ-014-03-20",
                      original_contract_value=Decimal("0"), revised_total=Decimal("1000.04"),
                      approval_status="PENDING", line_items=items)
    try:
        ChangeOrderSchema(change_order_number="CO-2024-090", wbs_code="PROJ-014-03-20",
                          original_contract_value=Decimal("0"), revised_total=Decimal("1000.06"),
                          approval_status="PENDING", line_items=items)
        assert False, "expected ValueError"
    except ValidationError:
        pass

def test_low_confidence_quarantines_without_guessing():
    fields = [FieldExtraction("revised_total", None, 0.40, raw_crop_ref="s3://crops/co90.png")]
    out = classify_extraction(fields)
    assert out["state"] == "QUARANTINE"
    assert out["error_code"] == "OCR_UNRESOLVED_FIELD"
    assert out["raw_crop_ref"]  # crop preserved for re-capture

def test_routing_channel_by_prefix():
    assert determine_routing_channel("CALCULATION_MISMATCH") == AlertChannel.ESTIMATOR
    assert determine_routing_channel("OCR_UNRESOLVED_FIELD") == AlertChannel.DOC_CONTROL
    assert determine_routing_channel("TRANSPORT_GATEWAY_TIMEOUT") == AlertChannel.DEV_OPS

Run the suite with python -m pytest tests/test_error_handling.py -v. A green run proves the four error branches each emit a typed payload with a traceable hash and never fabricate a financial value.

Troubleshooting

OCR confidence collapses on stamped drawings. A revision stamp or wet signature overlapping a total cell drags the bounding-box confidence below 0.75, sending good documents to quarantine en masse. Root cause: the segmentation step treats stamp ink as field content. Fix: add a stamp/signature mask in OCR preprocessing before extraction so the overlay is removed rather than read, then re-score.

Pydantic regex mismatch on European decimal formats. A subcontractor submits 1.250,00 and Decimal coercion raises InvalidOperation, returning SCHEMA_TYPE_MISMATCH for what is actually valid data. Root cause: the schema assumes US ,/. conventions. Fix: normalize numeric strings in the extraction layer (strip thousands separators, standardize the decimal point) before they reach the schema, and keep the raw string in the audit trail.

Calculation mismatch on every retention-bearing change order. Totals fail the tolerance check because retention or tax lines are excluded from the line-item sum. Root cause: the cross-field validator sums only base line items. Fix: model retention and tax as explicit line items or extend the validator to apply the contractual multiplier before comparing to revised_total.

Alert storm during the monthly pay cycle. A burst of change orders trips dozens of CALCULATION_MISMATCH alerts and pages the estimator continuously. Root cause: dispatch is per-document with no batching. Fix: route alerts through the async queue with windowed aggregation, and rate-limit per channel so a group of related failures collapses into one digest.

Documents vanish after repeated transport failures. A webhook outage exhausts retries and the payload is lost. Root cause: the circuit-breaker path drops messages instead of parking them. Fix: on exhausted retries, push the full exception payload (original file, extraction output, context) to the dead-letter queue so it is replayable once the gateway recovers — a transport fault should cause a delay, not permanent data loss.

Frequently Asked Questions

Related

• Implementing retry logic for failed API document pulls
• Schema Validation Rules
• Field Extraction Techniques
• OCR Preprocessing for Construction Docs
• Fallback Alert Routing

← Back to Automated Document Ingestion & Parsing