Error Categorization & Logging

Architectural Boundary & Control Plane Mandate #

Within the broader Data Ingestion & Grant Parsing Workflows architecture, Error Categorization & Logging operates as a discrete, read-only control plane. Its sole mandate is to intercept, classify, and route pipeline anomalies without mutating upstream payloads or interrupting downstream reconciliation. This subsystem functions as a hard enforcement layer: it validates structural integrity, enforces semantic boundaries, and guarantees immutable auditability. Nonprofit operations teams, grant managers, Python automation developers, and compliance officers must treat this stage as a deterministic routing switch, not a corrective or transformational engine.

The control plane enforces strict separation of concerns. Parsing, normalization, and financial aggregation occur in adjacent stages. This boundary only evaluates compliance, captures failure signatures, writes to a write-ahead log (WAL), and routes records to deterministic fallback queues. No heuristic correction, auto-repair, or inline mutation is permitted.

Error Taxonomy & Deterministic Routing Matrix #

All pipeline exceptions must conform to a strict, enumerated taxonomy. Categories are defined by failure origin, compliance impact, and required routing behavior. The taxonomy eliminates ambiguity and ensures consistent handling across distributed batch workers.

Routing decisions are deterministic and version-controlled. INFRASTRUCTURE_TRANSIENT errors trigger isolated retry queues with capped exponential backoff. STRUCTURAL_PARSE and SEMANTIC_VALIDATION failures immediately quarantine the record and emit a structured alert to the operations dashboard. COMPLIANCE_RULE violations halt downstream propagation and require explicit compliance officer sign-off before any further processing.

Canonical Logging Architecture & Write-Ahead Enforcement #

Implementation relies on canonical Python tooling to guarantee deterministic output and machine-readable auditability. The standard logging module must be wrapped with structlog to enforce JSON-formatted log lines. Every log entry adheres to a fixed schema: timestamp, pipeline_stage, error_category, record_hash, validation_rule_id, fallback_action, and compliance_flag.

Deterministic record hashing (e.g., hashlib.sha256 applied to the canonicalized payload) ensures traceability across retries and prevents duplicate alerting. The logging pipeline must operate independently of business logic, writing to a write-ahead log (WAL) before any fallback routing executes. This guarantees that even catastrophic process failures preserve a complete, queryable audit trail for regulatory review.

import hashlib
import json
from enum import Enum
from typing import Any, Dict
from pathlib import Path
import structlog
from structlog.stdlib import BoundLogger
from pydantic import ValidationError

# ---------------------------------------------------------------------------
# Canonical Taxonomy & Routing
# ---------------------------------------------------------------------------
class ErrorCategory(str, Enum):
    STRUCTURAL_PARSE = "STRUCTURAL_PARSE"
    SEMANTIC_VALIDATION = "SEMANTIC_VALIDATION"
    COMPLIANCE_RULE = "COMPLIANCE_RULE"
    INFRASTRUCTURE_TRANSIENT = "INFRASTRUCTURE_TRANSIENT"

class RoutingAction(str, Enum):
    QUARANTINE = "QUARANTINE"
    RETRY_BACKOFF = "RETRY_BACKOFF"
    COMPLIANCE_HOLD = "COMPLIANCE_HOLD"

# ---------------------------------------------------------------------------
# Structlog Configuration (JSON-only, ISO timestamps, strict schema)
# ---------------------------------------------------------------------------
structlog.configure(
    processors=[
        structlog.stdlib.add_log_level,
        structlog.stdlib.add_logger_name,
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.processors.JSONRenderer()
    ],
    wrapper_class=structlog.stdlib.BoundLogger,
    context_class=dict,
    logger_factory=structlog.stdlib.LoggerFactory(),
)

# ---------------------------------------------------------------------------
# WAL Writer (Decoupled from routing logic)
# ---------------------------------------------------------------------------
class AuditWAL:
    def __init__(self, wal_path: str = "/var/log/grant_pipelines/error_wal.jsonl"):
        self.path = Path(wal_path)
        self.path.parent.mkdir(parents=True, exist_ok=True)
        self._logger: BoundLogger = structlog.get_logger("audit_wal")

    def write(self, entry: Dict[str, Any]) -> None:
        """Append to WAL synchronously before any routing occurs."""
        with open(self.path, "a", encoding="utf-8") as f:
            f.write(json.dumps(entry, sort_keys=True) + "\n")
        self._logger.info("wal_entry_persisted", record_hash=entry.get("record_hash"))

# ---------------------------------------------------------------------------
# Deterministic Routing Engine
# ---------------------------------------------------------------------------
def classify_and_route(
    payload: Dict[str, Any],
    schema_version: str,
    wal: AuditWAL
) -> RoutingAction:
    """
    Validates payload against frozen schema, logs deterministically,
    and returns routing action. NO MUTATION OCCURS.
    """
    record_hash = hashlib.sha256(
        json.dumps(payload, sort_keys=True).encode()
    ).hexdigest()
    
    logger = structlog.get_logger().bind(
        pipeline_stage="error_categorization",
        record_hash=record_hash,
        validation_rule_id=schema_version
    )

    try:
        # GrantRecord is a versioned Pydantic model defined in the validation boundary
        from validation_schemas import GrantRecord
        GrantRecord.model_validate(payload)
        return RoutingAction.QUARANTINE  # Default safe state if validation passes unexpectedly
    except ValidationError as exc:
        # Map Pydantic errors to canonical taxonomy
        category = _map_validation_error_to_category(exc)
        action = _map_category_to_action(category)
        compliance_flag = category == ErrorCategory.COMPLIANCE_RULE

        audit_entry = {
            "timestamp": structlog.contextvars.get_contextvars().get("timestamp", ""),
            "pipeline_stage": "error_categorization",
            "error_category": category.value,
            "record_hash": record_hash,
            "validation_rule_id": schema_version,
            "fallback_action": action.value,
            "compliance_flag": compliance_flag,
            "raw_error_signature": str(exc)
        }
        
        # WAL write precedes routing
        wal.write(audit_entry)
        
        logger.error(
            "validation_boundary_breach",
            error_category=category.value,
            fallback_action=action.value,
            compliance_flag=compliance_flag
        )
        return action

def _map_validation_error_to_category(exc: ValidationError) -> ErrorCategory:
    """Deterministic mapping from Pydantic errors to taxonomy."""
    for err in exc.errors():
        if "type" in err and "value_error" in err["type"]:
            return ErrorCategory.SEMANTIC_VALIDATION
        if "type" in err and "missing" in err["type"]:
            return ErrorCategory.STRUCTURAL_PARSE
    return ErrorCategory.SEMANTIC_VALIDATION

def _map_category_to_action(category: ErrorCategory) -> RoutingAction:
    mapping = {
        ErrorCategory.INFRASTRUCTURE_TRANSIENT: RoutingAction.RETRY_BACKOFF,
        ErrorCategory.COMPLIANCE_RULE: RoutingAction.COMPLIANCE_HOLD,
        ErrorCategory.STRUCTURAL_PARSE: RoutingAction.QUARANTINE,
        ErrorCategory.SEMANTIC_VALIDATION: RoutingAction.QUARANTINE,
    }
    return mapping[category]

Validation Boundaries & Fallback Routing Implementation #

Explicit validation occurs at the subsystem boundary. Validation rules are declarative, version-controlled, and evaluated against a frozen schema snapshot using pydantic or cerberus. When a record fails validation, the system must not attempt inline correction. Instead, it applies a deterministic fallback action based on the routing matrix.

The fallback routing engine operates as a state machine. Records are never modified in place. Instead, the original payload is serialized alongside its routing directive and dispatched to isolated message queues (e.g., AWS SQS, RabbitMQ, or Redis Streams). This design ensures that downstream consumers can process quarantined records, retry transient failures, or route compliance holds without risking data corruption or audit trail fragmentation.

Key implementation constraints:

1. Immutable Payloads: The original byte stream or parsed dictionary is preserved exactly as received.
2. Schema Versioning: Validation rules are tied to a validation_rule_id (e.g., v2.1.0). Schema drift triggers STRUCTURAL_PARSE classification automatically.
3. Bounded Retries: INFRASTRUCTURE_TRANSIENT retries are capped at 3 attempts with exponential backoff. After exhaustion, records transition to QUARANTINE.
4. Idempotent Dispatch: Routing actions are idempotent. Duplicate payloads with identical record_hash values are deduplicated at the queue ingress layer.

Compliance Mapping & Regulatory Traceability #

The error categorization subsystem directly maps to nonprofit financial compliance frameworks. Every log entry and routing decision serves as auditable evidence for internal controls and external examinations.

Compliance officers can query the WAL using standard JSON path filters to reconstruct the exact state of any grant record at the moment of failure. The compliance_flag boolean enables rapid filtering for regulatory submissions. All logs are retained for a minimum of seven years, aligned with federal grant record retention policies.

Pipeline Handoffs & Operational Integration #

This control plane interfaces with upstream and downstream stages exclusively through well-defined handoff boundaries. It does not perform parsing, normalization, or financial aggregation.

• Ingestion Handoff: Raw outputs from PDF Grant Application Parsing are passed as canonical dictionaries. Structural anomalies in OCR extraction or table boundary detection are immediately classified as STRUCTURAL_PARSE and quarantined.
• Financial Sync Handoff: Normalized rows from Excel Budget Template Sync undergo semantic validation. Out-of-range indirect cost rates or mismatched cost center codes trigger SEMANTIC_VALIDATION routing without altering the original spreadsheet payload.
• Operational Monitoring: Aggregated error streams feed into centralized observability dashboards. For production deployment patterns, alert routing, and log aggregation strategies, refer to Implementing automated error logging for grant pipelines.

Operations teams must configure alert thresholds based on error category volume. A spike in COMPLIANCE_RULE violations requires immediate workflow suspension and manual triage. INFRASTRUCTURE_TRANSIENT spikes should trigger infrastructure scaling or rate-limit renegotiation. The subsystem guarantees that no malformed or non-compliant record silently propagates to financial reconciliation or funder reporting stages.