Debugging JSON Schema Validation Failures on CMMS Work Order Payloads

Automated preventive maintenance routing stalls at the validation gate when generated work orders are submitted to a CMMS ingestion endpoint and the draft-07 validator rejects payloads that look structurally sound. Two failures dominate the incident logs: a string-formatted PM interval that violates a strict integer type, and a missing parent_asset_id rejected by an unconditional required array on a nested asset object. Both are surfaced by the same validation stage defined in the work order schema standards, and both are fixable without watering down the contract. This page isolates the exact log trace, explains the root cause, and gives you a runnable before/after fix.

Incident Profile

When a Python automation script dispatches a batch of generated work orders, the validation layer returns 400 Bad Request with a structured error body. The application log captures the precise jsonschema exception chain:

2026-06-14 08:12:03,441 [ERROR] cmms.validation.gateway: Payload validation failed for WO-8842
Traceback (most recent call last):
  File "/opt/cmms-integration/validators/schema_engine.py", line 42, in validate_payload
    jsonschema.validate(instance=payload, schema=WORK_ORDER_SCHEMA)
  File "/usr/local/lib/python3.11/site-packages/jsonschema/validators.py", line 1332, in validate
    raise error
jsonschema.exceptions.ValidationError: '30d' is not of type 'integer'

Failed validating 'type' in schema['properties']['pm_interval']:
    {'type': 'integer', 'minimum': 1, 'description': 'Days between PM executions'}

On instance['pm_interval']:
    '30d'

During handling of the above exception, another exception occurred:

jsonschema.exceptions.ValidationError: 'parent_asset_id' is a required property

Failed validating 'required' in schema['properties']['asset_hierarchy']:
    {'type': 'object', 'required': ['parent_asset_id', 'location_code', 'criticality_score']}

On instance['asset_hierarchy']:
    {'location_code': 'BLDG-A-04', 'criticality_score': 3}

The pipeline halts mid-batch. Maintenance engineers see delayed PM execution, and integration teams watch the broker backlog climb because every rejected message blocks the slot behind it. The symptom that distinguishes this incident from a generic 400 is the paired traceback: a type violation on pm_interval and a required violation on asset_hierarchy, raised against payloads that round-trip cleanly through every other stage.

Root Cause Analysis

Two independent contract violations compound into a single rejection, and each originates outside the validator itself.

  1. Type coercion mismatch on pm_interval. The upstream ERP or scheduling engine exports interval values as strings with unit suffixes ("30d", "2w"), a format that traces back to how PM cadences are stored before PM interval calculation resolves them to calendar days. The CMMS schema enforces a strict integer representing days. The jsonschema library performs no implicit coercion, so "30d" is rejected on sight. The correct boundary for coercion is the normalization step, never the validator — a validator that “helpfully” casts types hides drift instead of failing loudly.

  2. Unconditional required array on asset_hierarchy. The schema mandates parent_asset_id on every asset node. But root-level equipment and standalone assets have no parent, which is a legitimate state in any tree produced by asset hierarchy design. Listing parent_asset_id in an unconditional required array treats a valid root node as a hard error. The contract needs conditional logic: enforce parent_asset_id only when the node declares itself a child.

Together these indicate a misalignment between the upstream data-generation contract and the strict validation boundary. The fix is two-sided — normalize the payload before validation, and correct the schema so it models the real asset topology.

Resolution

Fix 1 — coerce intervals at the normalization boundary

The string-to-integer conversion belongs in a dedicated normalizer that runs before the payload ever reaches the validator. It parses the unit suffix, converts to days, and raises a typed error on anything it cannot recognize.

import re

# AFTER: explicit coercion at the normalization boundary, never inside the validator
def parse_pm_interval(interval: object) -> int:
    """Coerce '30d', '2w', '1m', '1y' (or a bare int) into integer days."""
    if isinstance(interval, int):
        return interval
    match = re.match(r"^(\d+)([dwmy])$", str(interval).strip(), re.IGNORECASE)
    if not match:
        raise ValueError(f"Unrecognized PM interval format: {interval!r}")
    value, unit = int(match.group(1)), match.group(2).lower()
    multipliers = {"d": 1, "w": 7, "m": 30, "y": 365}
    return value * multipliers[unit]

Fix 2 — make parent_asset_id conditionally required

The broken schema declared the field unconditionally. The corrected version uses draft-07 if/then: the node is only forced to carry a parent_asset_id when it advertises itself as a child via node_type.

# BEFORE: every asset node must carry a parent — rejects valid root assets
BROKEN_HIERARCHY = {
    "type": "object",
    "required": ["parent_asset_id", "location_code", "criticality_score"],
}

# AFTER: parent_asset_id is required only when node_type == "child"
FIXED_HIERARCHY = {
    "type": "object",
    "properties": {
        "node_type": {"enum": ["root", "child"]},
        "parent_asset_id": {"type": "string", "minLength": 1},
        "location_code": {"type": "string", "pattern": "^[A-Z]+-[A-Z]-[0-9]{2}$"},
        "criticality_score": {"type": "integer", "minimum": 1, "maximum": 5},
    },
    "required": ["node_type", "location_code", "criticality_score"],
    "if": {"properties": {"node_type": {"const": "child"}}},
    "then": {"required": ["parent_asset_id"]},
}

The difference is behavioural: a root chiller now validates without a parent, while a child pump that omits its parent_asset_id still fails loudly — which is exactly the orphan-reference case you want caught before routing.

Minimal Reproducible Pipeline

The script below is end-to-end runnable. It imports the canonical WorkOrderPayload so the validated record is the exact object the routing engine, the PM scheduler, and the parts-reservation stage all consume, normalizes the interval, validates against the corrected contract, and constructs the typed payload. The SLA fields priority, requested_completion, and escalation_tier are mandatory on every work order on this site, so they are carried through here too.

import re
from dataclasses import dataclass, field
from datetime import datetime, timezone
from enum import Enum
from typing import Dict, List, Optional

from jsonschema import Draft7Validator


class Priority(str, Enum):
    CRITICAL = "critical"
    HIGH = "high"
    STANDARD = "standard"
    PLANNED = "planned"


class PMTriggerType(str, Enum):
    CALENDAR = "calendar"
    RUNTIME_HOURS = "runtime_hours"
    CYCLE_COUNT = "cycle_count"
    CONDITION_BASED = "condition_based"


@dataclass(frozen=True)
class WorkOrderPayload:
    """Canonical CMMS work order — SLA fields are mandatory site-wide."""
    work_order_id: str
    asset_id: str
    required_skill_codes: List[str]
    location_zone: str
    part_skus: List[str] = field(default_factory=list)
    required_quantities: Dict[str, int] = field(default_factory=dict)
    pm_trigger_type: Optional[PMTriggerType] = None
    routing_policy_id: str = "skill_weighted"
    priority: Priority = Priority.STANDARD
    requested_completion: Optional[datetime] = None
    escalation_tier: int = 0
    status: str = "open"
    created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))


class SchemaRejection(ValueError):
    """Raised when a payload fails the draft-07 contract; names the failed field."""
    def __init__(self, failed_field: str, message: str):
        self.failed_field = failed_field
        super().__init__(message)


WORK_ORDER_SCHEMA = {
    "$schema": "http://json-schema.org/draft-07/schema#",
    "type": "object",
    "properties": {
        "work_order_id": {"type": "string", "minLength": 1},
        "asset_id": {"type": "string", "minLength": 1},
        "pm_interval": {"type": "integer", "minimum": 1},
        "priority": {"enum": [p.value for p in Priority]},
        "escalation_tier": {"type": "integer", "minimum": 0, "maximum": 3},
        "asset_hierarchy": {
            "type": "object",
            "properties": {
                "node_type": {"enum": ["root", "child"]},
                "parent_asset_id": {"type": "string", "minLength": 1},
                "location_code": {"type": "string", "pattern": "^[A-Z]+-[A-Z]-[0-9]{2}$"},
                "criticality_score": {"type": "integer", "minimum": 1, "maximum": 5},
            },
            "required": ["node_type", "location_code", "criticality_score"],
            "if": {"properties": {"node_type": {"const": "child"}}},
            "then": {"required": ["parent_asset_id"]},
        },
    },
    "required": ["work_order_id", "asset_id", "pm_interval", "priority", "asset_hierarchy"],
    "additionalProperties": False,
}


def parse_pm_interval(interval: object) -> int:
    """Coerce '30d', '2w', '1m', '1y' (or a bare int) into integer days."""
    if isinstance(interval, int):
        return interval
    match = re.match(r"^(\d+)([dwmy])$", str(interval).strip(), re.IGNORECASE)
    if not match:
        raise ValueError(f"Unrecognized PM interval format: {interval!r}")
    value, unit = int(match.group(1)), match.group(2).lower()
    multipliers = {"d": 1, "w": 7, "m": 30, "y": 365}
    return value * multipliers[unit]


def normalize_and_validate(raw: dict) -> WorkOrderPayload:
    """Coerce types, validate against the corrected contract, return a typed payload."""
    # Step A: coerce the interval before it reaches the validator
    if "pm_interval" in raw:
        raw["pm_interval"] = parse_pm_interval(raw["pm_interval"])

    # Step B: collect every error, then raise the deepest one with its field name
    validator = Draft7Validator(WORK_ORDER_SCHEMA)
    errors = sorted(validator.iter_errors(raw), key=lambda e: list(e.absolute_path))
    if errors:
        first = errors[0]
        field_name = first.absolute_path[-1] if first.absolute_path else "(root)"
        raise SchemaRejection(str(field_name), first.message)

    # Step C: construct the immutable payload downstream stages consume
    return WorkOrderPayload(
        work_order_id=raw["work_order_id"],
        asset_id=raw["asset_id"],
        required_skill_codes=raw.get("required_skill_codes", ["GEN-01"]),
        location_zone=raw["asset_hierarchy"]["location_code"],
        priority=Priority(raw["priority"]),
        escalation_tier=raw.get("escalation_tier", 0),
        pm_trigger_type=PMTriggerType.CALENDAR,
    )


if __name__ == "__main__":
    # Previously rejected: string interval + a root asset with no parent
    payload = {
        "work_order_id": "WO-8842",
        "asset_id": "f47ac10b-58cc-4372-a567-0e02b2c3d479",
        "pm_interval": "30d",
        "priority": "high",
        "escalation_tier": 1,
        "asset_hierarchy": {
            "node_type": "root",
            "location_code": "BLDG-A-04",
            "criticality_score": 3,
        },
    }
    try:
        work_order = normalize_and_validate(payload)
        print(f"validated {work_order.work_order_id} | priority:{work_order.priority.value}")
    except SchemaRejection as exc:
        print(f"rejected on field '{exc.failed_field}': {exc}")

Running it prints validated WO-8842 | priority:high. The "30d" interval is now coerced to 30, and the root asset validates because node_type is root. Switch node_type to child without adding a parent_asset_id and the same script raises rejected on field 'asset_hierarchy' — the orphan guard you actually want.

Before and after: the same payload at the validation gate Two parallel paths for an identical work order payload that carries a string pm_interval of 30d and a root asset node with no parent. On the left, the broken path sends the raw payload straight into an unconditional required gate; because pm_interval is a string and parent_asset_id is missing, the gate raises a draft-07 type error and required error, and the message is dropped into a dead-letter queue, blocking the batch. On the right, the fixed path first normalizes the payload, coercing the string 30d into the integer 30, then validates against a conditional draft-07 gate whose rule only requires parent_asset_id when node_type equals child; the root node passes, and a typed WorkOrderPayload is emitted to the routing engine. One payload, two outcomes at the draft-07 gate BEFORE — strict schema, raw input AFTER — normalize, then validate pm_interval:"30d" · asset_hierarchy:{node_type:"root"} pm_interval:"30d" · asset_hierarchy:{node_type:"root"} Raw payload submitted straight to the validator Unconditional required gate type: integer · required: [parent_asset_id, …] no coercion, no conditional logic 2 ValidationErrors '30d' is not of type 'integer' 'parent_asset_id' is a required property — but the root node is legitimately parentless Dead-letter queue batch slot blocked behind it Raw payload enters the normalization boundary first Normalize parse_pm_interval("30d") → 30 Conditional draft-07 gate if node_type=="child" then required:["parent_asset_id"] root passes · orphan child still fails loudly Typed WorkOrderPayload emitted → routing engine The contract stays strict either way — the fix moves coercion to the boundary and models the real asset topology.

Prevention Checklist

Work through these to stop the failure recurring. The items render as interactive checkboxes.

Frequently Asked Questions

Why not let the validator coerce string intervals automatically?

Because a validator that casts types hides the drift that caused the problem. If "30d" is silently accepted today, the day the upstream format changes to "30 days" you get a wrong answer instead of a clean rejection. Coerce explicitly in the normalizer where the rule is visible and testable, and keep the validator strict.

Does upgrading from draft-07 to 2020-12 change this fix?

Yes — if/then/else evaluation and $ref resolution behave differently across drafts. Pin the validator to a specific draft and test the conditional required branch before migrating, because a draft change can quietly turn a child node’s missing parent_asset_id from a rejection into a pass.

Where should this validation run in the pipeline?

Synchronously at the boundary, before the payload consumes a worker slot. The validator is stateless and idempotent, so it is cheap to run inline; only the publish step needs the retry and dead-letter machinery, which keeps a single malformed work order from stalling the whole batch.

Anchor every parent_asset_id against the equipment tree with asset hierarchy design, source interval values from PM interval calculation, and feed validated payloads downstream through async batch processing.

Part of: Work Order Schema Standards.