Policy Schema Design for Insurance Claims & Policy Data Automation

Policy schema design is the typed contract that every other component in the Core Architecture & Compliance Mapping domain depends on — the single definition of what a valid policy record is before any claim is allowed to act on it. This component sits directly beneath that parent domain and defines the field-level constraints, coverage enumerations, and jurisdictional flags that downstream automation treats as ground truth.

A production policy schema is not documentation. It is executable validation that bridges legacy underwriting cores, modern claims automation, and statutory compliance into one deterministic boundary. Rigorously typed schemas eliminate ambiguity, force every record onto a single predictable routing path, and reject malformed payloads at the edge rather than letting them corrupt adjudication days later. When the schema is loose, the failures are quiet: a coverage limit parsed as a string instead of a decimal, a missing effective-date window that lets an expired policy authorize a payout, a state code that never gets validated and routes a Florida claim through Texas rules.

At pilot volume a few hundred clean policies a day move through hand-written dict access and a handful of if checks without obvious problems. The failure classes only surface once the platform ingests hundreds of thousands of heterogeneous records from legacy administration exports, third-party administrator feeds, and partner API webhooks.

The first failure is silent type coercion drift. A monetary limit arriving as "50000.00" from one carrier and 50000 from another, parsed loosely, produces inconsistent reserve math downstream. Float arithmetic on currency compounds the error until a reconciliation report disagrees with the ledger and no one can say which record introduced the discrepancy.

The second is non-deterministic routing from optional fields. When the schema permits a missing status or an unconstrained state_code, the triage layer has to guess, and guesses are not reproducible. The same policy re-validated during a replay can route differently than it did originally — fatal in a domain where every routing decision must be reconstructable during a market-conduct exam.

The third is boundary bleed across mutable records. When the validated policy object is a plain mutable dict shared across stages, a normalization step writes back over the raw intake, third-party data leaks into a field feeding an automated decision, and the audit trail records a record that no longer matches what was actually processed. A disciplined schema produces frozen, copy-across-boundary records so the validated artifact can never be silently mutated after the fact — the same discipline Data Boundary Enforcement enforces across the wider domain.

The schema’s job is to collapse all three risks into a single validation gate: coerce every field to an explicit type, reject anything that violates a constraint with a structured error, and emit an immutable, fully scoped record that the Claims Lifecycle Architecture triage router can consume without re-checking anything.

This component targets Python 3.10+ for modern type-hint syntax and frozen-dataclass ergonomics, and uses Pydantic v2 for runtime validation and structured error reporting. Pin the validation and serialization libraries explicitly so a minor version bump cannot silently change coercion or hashing semantics:

Copy
python -m venv .venv && source .venv/bin/activate
pip install "pydantic==2.*" "orjson==3.*"

Two infrastructure dependencies underpin schema operation in production. The first is a versioned schema registry that stores every contract revision keyed by a version string, so a policy validated last year can be re-validated against the exact constraints in force at the time. The second is the shared append-only audit store — the governance backbone common to the whole Core Architecture & Compliance Mapping domain — into which every rejection is written with its payload hash before the record is dropped.

Establish validation thresholds as environment-driven configuration before writing any logic: the active contract version (POLICY_SCHEMA_VERSION), the permitted currency set, and the per-line-of-business coverage minimums. None of these belong hard-coded inside the model; they are injected so that a tuning change is a configuration deploy, not a code edit.

Every payload, regardless of source, enters one validation gate and undergoes a fixed evaluation sequence: jurisdiction validation → coverage applicability verification → temporal window confirmation → status resolution. The order is deliberate. Jurisdiction resolves first because state code selects the rule set every later check depends on; coverage applicability runs before temporal checks so a policy with no routable coverage is rejected cheaply; the effective/expiration window is confirmed before status so an “active” flag on an expired policy cannot slip through. This ordered evaluation guarantees that the triage engine receives a correctly scoped policy context regardless of ingestion source.

The hard isolation rule is that the validated record is immutable. The schema emits frozen structures so no downstream stage can mutate a coverage limit or a jurisdiction after validation — the type system enforces the boundary rather than relying on convention. Where the policy data originates from scanned declarations rather than structured feeds, Policy PDF Parsing & Extraction Workflows produces the raw payload and Field Mapping Strategies normalize carrier-specific identifiers into the canonical field names this schema expects before the payload ever reaches the gate.

The validated record is the contract consumed downstream. The deterministic routing tier it carries feeds the Claims Triage & Routing Engines control plane, and the coverage enumerations it locks in are exactly what Coverage Validation Rules check a loss against during adjudication. Nothing downstream re-parses the policy; it trusts the schema.

The model below demonstrates strict typing, a fixed evaluation order via ordered validators, frozen records that enforce immutability, deterministic routing-tier resolution, and structured error capture with a payload hash for the audit trail. It uses Pydantic v2’s validation model and Python’s decimal module for exact monetary arithmetic — never float — so reserve math is reproducible to the cent.

Copy
from __future__ import annotations

import hashlib
import json
import logging
from datetime import date
from decimal import Decimal
from typing import List, Literal, Union

from pydantic import (
    BaseModel,
    ConfigDict,
    Field,
    ValidationError,
    field_validator,
    model_validator,
)

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s | %(name)s | %(levelname)s | %(message)s",
)
logger = logging.getLogger(__name__)


class CoverageLimit(BaseModel):
    model_config = ConfigDict(frozen=True)
    limit_type: Literal["per_occurrence", "aggregate", "split"]
    amount: Decimal = Field(gt=0, description="Positive, non-zero monetary value")
    currency: Literal["USD", "CAD"] = "USD"


class JurisdictionData(BaseModel):
    model_config = ConfigDict(frozen=True)
    state_code: str = Field(pattern=r"^[A-Z]{2}$", description="ISO 3166-2:US state code")
    regulatory_version: str = Field(min_length=3, max_length=10)


class PolicySchema(BaseModel):
    # frozen -> no downstream stage can mutate a validated record;
    # extra="forbid" -> an unexpected field is a rejection, not a silent pass-through.
    model_config = ConfigDict(frozen=True, extra="forbid")
    policy_id: str = Field(min_length=10, max_length=20, description="Immutable policy identifier")
    effective_date: date
    expiration_date: date
    jurisdiction: JurisdictionData
    status: Literal["active", "suspended", "cancelled", "expired"]
    coverage_limits: List[CoverageLimit]

    @model_validator(mode="before")
    @classmethod
    def validate_temporal_window(cls, data: Union[dict, object]) -> Union[dict, object]:
        if isinstance(data, dict):
            eff = data.get("effective_date")
            exp = data.get("expiration_date")
            if eff and exp and eff >= exp:
                raise ValueError("Expiration date must strictly follow effective date")
        return data

    @field_validator("coverage_limits", mode="after")
    @classmethod
    def enforce_routable_coverage(cls, limits: List[CoverageLimit]) -> List[CoverageLimit]:
        if not limits:
            raise ValueError("At least one coverage limit is required for routing")
        return limits


def route_policy_payload(raw_payload: dict, schema_version: str) -> dict:
    """
    Validation-gate entry point. Validates a raw payload from any source,
    enforces the schema boundary, and returns a deterministic routing
    instruction for the triage engine, or a structured rejection for audit.
    """
    try:
        policy = PolicySchema.model_validate(raw_payload)
        logger.info("Policy %s validated against schema %s.", policy.policy_id, schema_version)

        has_per_occurrence = any(
            c.limit_type == "per_occurrence" for c in policy.coverage_limits
        )
        routing_tier = "priority" if has_per_occurrence else "standard"

        return {
            "status": "routed",
            "policy_id": policy.policy_id,
            "jurisdiction": policy.jurisdiction.state_code,
            "schema_version": schema_version,
            "routing_tier": routing_tier,
            "compliance_flags": [],
        }
    except ValidationError as exc:
        canonical = json.dumps(raw_payload, sort_keys=True, separators=(",", ":"), default=str)
        payload_hash = hashlib.sha256(canonical.encode("utf-8")).hexdigest()
        logger.error("Schema %s validation failed for hash=%s", schema_version, payload_hash)
        return {
            "status": "rejected",
            "schema_version": schema_version,
            "errors": exc.errors(),
            "raw_payload_hash": payload_hash,
            "compliance_flags": ["MALFORMED_PAYLOAD"],
        }

Two design points make this gate auditable. First, route_policy_payload never raises out of the function — a malformed payload is degraded to a structured rejected result carrying the exact field paths from exc.errors() and a SHA-256 hash of the canonically-serialized payload, so a rejection is fully reconstructable without ever persisting raw, possibly-sensitive policy data. Second, every result is stamped with the schema_version that produced it, so a record validated under an older contract can be re-evaluated against the exact constraints in force at the time rather than today’s. The frozen=True configuration on all three models is what guarantees the validated record cannot be mutated after the gate; extra="forbid" turns an unexpected field into a rejection instead of a silent pass-through that hides a mapping drift.

The model holds no business thresholds of its own beyond structural constraints. Coverage minimums, permitted currencies, and the active contract version are injected at load time so a tuning change is a configuration deploy rather than a code edit. Carrier- and state-specific overrides should be resolved when the schema is constructed, never at validation time:

Copy
import os

SCHEMA_VERSION = os.environ.get("POLICY_SCHEMA_VERSION", "2026.06.0")

# Per-line-of-business coverage floors, sourced from the versioned rule store.
COVERAGE_MINIMUMS = {
    "commercial_auto": Decimal("50000"),
    "general_liability": Decimal("100000"),
}


def enforce_coverage_floor(policy: PolicySchema, line_of_business: str) -> list[str]:
    """Return compliance flags for any aggregate limit below the jurisdiction floor."""
    floor = COVERAGE_MINIMUMS.get(line_of_business)
    if floor is None:
        return []
    flags: list[str] = []
    for limit in policy.coverage_limits:
        if limit.limit_type == "aggregate" and limit.amount < floor:
            flags.append(f"BELOW_MIN_{line_of_business.upper()}")
    return flags

Because the version is captured in every routing result, a tuning change is a new schema document and a new version string — never an in-place edit — so the relationship between a validated record and the contract that produced it is always reconstructable. The per-state coverage minimums and mandatory-disclosure rules that feed COVERAGE_MINIMUMS derive from State Regulation Mapping, keeping jurisdictional values out of application code entirely. The coarse routing_tier the gate emits is later refined by the Automated Severity Scoring Models once a loss is attached to the policy.

Every validation outcome must map to a statutory mandate or an internal control, and the structured rejection record is the artifact that proves it did. By hashing the canonical payload at the gate and recording the schema_version applied, the component guarantees that any historical validation — pass or reject — can be reconstructed exactly: the same hash, the same contract version, the same outcome. This immutability is what satisfies NAIC model-regulation expectations and internal SOX controls, and it lets compliance reporting run directly off the audit store without reconciling against the live system.

Jurisdictional variation in mandatory disclosures, coverage minimums, and exclusion clauses is encoded directly into the validation layer rather than handled as after-the-fact reconciliation. The structured error serialization aligns with the JSON Schema specification’s validation-reporting model, exposing the exact field path and constraint violated without leaking sensitive values, so extraction workflows remain transparent and reproducible across distributed environments. Translating industry-standard form libraries into these typed contracts — including version-pinning every ISO revision against its JSON contract — is covered in depth in how to map ISO policy forms to JSON schemas.

Schema failures cluster into a small set of named scenarios, each with a deterministic diagnostic path and a code-level fix.

Schema design is the contract the rest of this domain attaches to. Its hardest sub-problem — reconciling the structural variability of ISO form libraries, jurisdictional endorsements, and legacy carrier modifications against a single deterministic JSON contract — is treated in depth in how to map ISO policy forms to JSON schemas, which expands the canonical-modeling and memory-optimization patterns sketched here into a full schema-registry workflow. Upstream, raw records arrive from Policy PDF Parsing & Extraction Workflows and are normalized by Field Mapping Strategies; the per-jurisdiction constraints come from State Regulation Mapping; and the immutability that protects the audit trail is enforced alongside Data Boundary Enforcement. Downstream, the validated record feeds the Claims Lifecycle Architecture triage router.

• Core Architecture & Compliance Mapping — the parent domain defining the contracts this schema enforces
• How to map ISO policy forms to JSON schemas — the schema-registry workflow for translating ISO form libraries into typed contracts
• Claims Lifecycle Architecture — the triage router that consumes the validated record this gate emits
• State Regulation Mapping — supplies the per-jurisdiction coverage minimums and disclosure rules
• Data Boundary Enforcement — the isolation discipline that keeps frozen records and the audit trail trustworthy