Which schema validation mismatches must be retained for future ingestion runs?

**Date of Record:** 2026 October 25 **Scope:** Applies when an agentic ingestion pipeline or LLM runner encounters external data that fails strict schema validation, type checking, or structural constraints, and is preparing to process subsequent batches. Does not apply to catastrophic parsing failures that abort the entire pipeline, or to dynamic schemas where structural variance is explicitly expected.

Core Distinction

The separation between **transient invalid payloads** (the raw, malformed data rejected during ingestion) and **schema mismatch signatures** (the deterministic diff between the expected contract and the observed malformation). The raw payload is garbage and pollutes memory if retained; the mismatch signature is a negative constraint that prevents a successor from wasting compute re-validating the same bad data. An ingestion pipeline that discards the raw payload without recording the signature will repeatedly fail on the same upstream drift.

What Was Observed

In data-processing agent architectures, failing to retain schema mismatch signatures produces three primary failure modes:

  1. **Redundant Rejection Loops:** A successor ingestion run pulls the same upstream batch, runs the same heavy validation logic, and rejects the exact same malformed records, wasting tokens and compute.
  2. **Silent Upstream Drift:** An API upstream changes a field type (e.g., `integer` to `string`), and the pipeline silently skips the malformed records. The instance never records the systemic drift, causing downstream analytics to confidently operate on incomplete datasets.
  3. **Over-Permissive Coercion:** An instance encounters a mismatch (e.g., `null` where a list was expected) and attempts to force-ingest the data by substituting an empty list. Without a record of this mitigation, the successor cannot distinguish between a genuinely empty list and a schema violation.

Recorded Elements

  1. **`mismatch_signature`:** The exact structural diff (e.g., `user.age: expected int, observed null` or `missing_required_key: transaction_id`), stripped of the actual data values.
  2. **`source_endpoint_identifier`:** The specific URI, API endpoint, or file path that supplied the malformed payload, enabling targeted upstream alerts.
  3. **`rejection_boundary_rule`:** The exact validation predicate that failed, preventing ambiguity about which constraint was violated.
  4. **`mitigation_directive`:** The deterministic action the successor must take upon re-encountering this signature (e.g., `SKIP_AND_ALERT`, `COERCE_TO_DEFAULT`, or `HALT_INGESTION`).

Condition of Validity

* **Applies when:** The ingestion pipeline validates data against a known schema, the data fails validation, and the pipeline will execute subsequent runs against the same upstream data source. * **Does not apply when:** The invalid payload was a one-off corrupted transmission (e.g., a truncated TCP packet) that will not recur, or the upstream source has been permanently decommissioned.

Decision Boundary

  1. **Did the payload fail schema validation?** If no, ingest normally. If yes, proceed to isolate the mismatch.
  2. **Is the mismatch structural (missing key, type collapse, constraint violation) rather than a transient network truncation?** If structural, extract the `mismatch_signature`. If transient, trigger a standard retry.
  3. **Is the exact schema path and observed type identifiable?** If yes, record the signature. If the payload is so malformed that a signature cannot be determined, log a `MALFORMED_PAYLOAD_HASH` instead to track repeated delivery of the same bad data.
  4. **Conversely, when should the mismatch be discarded rather than retained?** If the upstream source has been verified to return *only* this malformed data and has no valid records remaining, the mismatch record offers no future value and can be purged to save context.

Operational Effect

  1. **Compute Preservation:** Allows successors to drop known-bad payloads in O(1) time by matching their signatures against the negative constraint archive, avoiding expensive full-schema validation.
  2. **Drift Telemetry:** Surfaces silent upstream API changes as explicit operational records, enabling the agent to flag schema drift to human operators or adapter layers.
  3. **Deterministic Coercion:** Ensures that when an instance chooses to coerce malformed data to a valid state, that coercion logic is preserved and applied consistently by all future instances.
from dataclasses import dataclass
from enum import Enum, auto
import hashlib
import json


class MismatchSeverity(Enum):
    FATAL_HALT = auto()
    SKIP_AND_ALERT = auto()
    COERCIBLE = auto()


@dataclass(frozen=True)
class ValidationFailure:
    source_endpoint: str
    schema_path: str
    expected_type: str
    observed_type: str
    raw_payload_hash: str


@dataclass(frozen=True)
class MismatchRecord:
    source_endpoint: str
    schema_path: str
    expected_type: str
    observed_type: str
    severity: MismatchSeverity
    mitigation_directive: str


@dataclass(frozen=True)
class IngestionArchive:
    records: tuple[MismatchRecord, ...]
    archive_digest: str


class IngestionMismatchTracker:
    """
    Evaluates schema validation failures and seals negative constraint
    signatures for future ingestion runs, preventing redundant validation.
    """

    @staticmethod
    def _hash(data: str) -> str:
        return hashlib.sha256(data.encode("utf-8")).hexdigest()

    @classmethod
    def archive_failure(
        cls,
        failure: ValidationFailure,
        severity: MismatchSeverity,
    ) -> MismatchRecord:
        
        if not failure.source_endpoint.strip() or not failure.schema_path.strip():
            raise ValueError("Source endpoint and schema path are required for a mismatch record.")

        # Derive mitigation directive based on severity
        if severity is MismatchSeverity.FATAL_HALT:
            directive = f"HALT: Schema mismatch at {failure.schema_path} from {failure.source_endpoint}"
        elif severity is MismatchSeverity.SKIP_AND_ALERT:
            directive = f"SKIP: Ignore payload from {failure.source_endpoint}, alert on {failure.schema_path}"
        else:
            directive = f"COERCE: Substitute default for {failure.schema_path}"

        return MismatchRecord(
            source_endpoint=failure.source_endpoint,
            schema_path=failure.schema_path,
            expected_type=failure.expected_type,
            observed_type=failure.observed_type,
            severity=severity,
            mitigation_directive=directive,
        )

    @classmethod
    def build_archive(
        cls, records: tuple[MismatchRecord, ...]
    ) -> IngestionArchive:
        if not records:
            raise ValueError("Cannot build an empty ingestion archive.")

        digest_payload = {
            "records": tuple(
                sorted(
                    f"{r.source_endpoint}:{r.schema_path}:{r.observed_type}"
                    for r in records
                )
            )
        }
        digest = cls._hash(
            json.dumps(digest_payload, sort_keys=True, separators=(",", ":"))
        )

        return IngestionArchive(
            records=records,
            archive_digest=digest,
        )

    @classmethod
    def check_pre_flight(
        cls, archive: IngestionArchive, incoming_source: str, incoming_schema_path: str, incoming_type: str
    ) -> MismatchRecord | None:
        """
        Fetching-time check: returns a matching MismatchRecord if the incoming
        payload matches a known failure signature, bypassing full validation.
        """
        for record in archive.records:
            if (
                record.source_endpoint == incoming_source
                and record.schema_path == incoming_schema_path
                and record.observed_type == incoming_type
            ):
                return record
        
        return None