Work Order Schema Standards for the CMMS Routing Pipeline

Work order schema standards are the contract layer of the CMMS Architecture & Maintenance Taxonomy domain — the component that turns heterogeneous maintenance requests into a single typed payload the routing engine can dispatch deterministically.

When the schema is loose or implicit, the cost surfaces at the worst possible moment: a malformed request reaches the dispatch table, the routing engine cannot resolve a technician, and the task either stalls in a queue or fires against the wrong asset. Facilities managers lose the ability to prove SLA compliance, and integration teams absorb the reconciliation cost of every drifted field. Treating the schema as a strict, versioned boundary — validated once, immutable thereafter — is what lets preventive maintenance triggers translate directly into actionable dispatches. This guide implements the schema stage end to end: prerequisites, the input/output data contract, a step-by-step Python build, a configuration reference, validation checks, and the failure modes you will actually hit in production.

Prerequisites

The schema component runs at the seam between intake and routing. It consumes a normalized request envelope and emits a validated WorkOrderPayload; it never reaches into raw channel data or writes to dispatch tables directly. Before you deploy it, confirm the following are in place.

• Python 3.11+ with jsonschema>=4.21 for draft-07 contract enforcement and the standard-library dataclasses, enum, uuid, and datetime modules. No third-party scheduler is required because the validator is driven synchronously by the broker described in async batch processing.
• CMMS REST API v1 with write access to the work order resource (POST /api/v1/work-orders). The endpoint must reject unknown fields and return a structured 422 Unprocessable Entity body so validation failures are machine-readable rather than opaque.
• A resolved asset master. A work order’s asset_id is only meaningful once equipment is anchored to its functional location and parent-child dependencies, which is established through asset hierarchy design. An asset_id that does not resolve cannot be risk-weighted or routed to the correct crew.
• Environment variables: CMMS_BASE_URL, CMMS_API_TOKEN, and SCHEMA_STRICT_MODE (default true). The token must carry the workorders:write scope; a token missing it fails closed at the publish step rather than silently dropping the request.

Architecture and Data Contract

The component sits between a normalized request envelope and the routing engine. It runs four boundaries that keep the payload honest and stop routing logic from leaking upstream into the schema:

• Normalization boundary: raw inputs from intake adapters are mapped into the canonical field set; non-routing metadata is stripped so the payload stays lean.
• Validation boundary: the normalized payload is checked against a formal draft-07 contract. Same input, same verdict, every time — the stage is stateless and idempotent.
• Construction boundary: a valid payload is materialized into an immutable WorkOrderPayload dataclass that downstream stages consume without re-parsing.
• Publication boundary: the typed payload is handed to the routing engine, where it triggers skill matching, PM interval calculation precedence, and — when a job needs spares — parts availability checks. None of that routing is performed by this component.

The contract across the validation boundary is explicit. The input is a loosely typed dict (a normalized request envelope); the output is a WorkOrderPayload. Encoding the output as a frozen dataclass means a malformed request is rejected at the boundary instead of producing a plausible-but-wrong payload that silently mis-routes a task. The SLA fields — priority, requested_completion, and escalation_tier — are mandatory across every work order example on this site, because they are what the routing engine reads when it decides how urgently a task should dispatch and when it must escalate.

Step-by-Step Implementation

1. Define the canonical work order payload

Every component in the pipeline imports the same WorkOrderPayload rather than redefining its own shape, so a request validated here is the exact object the router, the PM scheduler, and the parts reservation stage all consume. This is the canonical definition; copy it verbatim into a shared module.

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


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))

2. Declare the validation contract

The wire-level contract is a draft-07 JSON schema. Keeping it separate from the dataclass lets the schema travel with the API gateway and lets non-Python consumers validate against the same source of truth. The additionalProperties: false constraint is what prevents schema drift — any field an upstream adapter invents is rejected rather than silently carried into routing.

WORK_ORDER_SCHEMA = {
    "$schema": "http://json-schema.org/draft-07/schema#",
    "type": "object",
    "required": [
        "work_order_id", "asset_id", "required_skill_codes",
        "location_zone", "priority",
    ],
    "properties": {
        "work_order_id": {"type": "string", "pattern": "^WO-[0-9]{4,}$"},
        "asset_id": {"type": "string", "format": "uuid"},
        "required_skill_codes": {
            "type": "array",
            "items": {"type": "string"},
            "minItems": 1,
        },
        "location_zone": {"type": "string", "pattern": "^[A-Z]{2,}-[0-9]{3}$"},
        "priority": {
            "enum": ["critical", "high", "standard", "planned"]
        },
        "requested_completion": {"type": "string", "format": "date-time"},
        "escalation_tier": {"type": "integer", "minimum": 0, "maximum": 3},
        "pm_trigger_type": {
            "enum": ["calendar", "runtime_hours", "cycle_count", "condition_based"]
        },
        "routing_policy_id": {"type": "string"},
        "part_skus": {"type": "array", "items": {"type": "string"}},
        "required_quantities": {
            "type": "object",
            "additionalProperties": {"type": "integer", "minimum": 1},
        },
    },
    "additionalProperties": False,
}

3. Normalize and validate the request

The validation stage strips non-routing metadata, coerces the asset_id to a canonical UUID string, checks the payload against the contract, and only then constructs the immutable dataclass. A rejection raises a structured error the caller can route to a dead-letter queue without guessing which field failed. The deeper edge cases — type coercion and nested asset mismatches — are covered in JSON schema validation for work order payloads.

import uuid
from datetime import datetime
from jsonschema import Draft7Validator


class SchemaRejection(ValueError):
    """Raised when a request fails the work order contract."""
    def __init__(self, failed_field: str, message: str):
        self.failed_field = failed_field
        self.message = message
        super().__init__(f"{failed_field}: {message}")


_VALIDATOR = Draft7Validator(WORK_ORDER_SCHEMA)


def normalize_and_validate(raw: dict) -> WorkOrderPayload:
    """Strip, coerce, validate, then construct an immutable payload."""
    # Keep only fields the contract knows about — defends against drift.
    clean = {k: v for k, v in raw.items() if k in WORK_ORDER_SCHEMA["properties"]}

    # Coerce asset_id to canonical UUID string before validation.
    if isinstance(clean.get("asset_id"), str):
        clean["asset_id"] = str(uuid.UUID(clean["asset_id"]))

    errors = sorted(_VALIDATOR.iter_errors(clean), key=lambda e: e.path)
    if errors:
        first = errors[0]
        field_name = first.path[-1] if first.path else first.validator
        raise SchemaRejection(str(field_name), first.message)

    completion = clean.get("requested_completion")
    return WorkOrderPayload(
        work_order_id=clean["work_order_id"],
        asset_id=clean["asset_id"],
        required_skill_codes=clean["required_skill_codes"],
        location_zone=clean["location_zone"],
        part_skus=clean.get("part_skus", []),
        required_quantities=clean.get("required_quantities", {}),
        pm_trigger_type=(
            PMTriggerType(clean["pm_trigger_type"])
            if clean.get("pm_trigger_type") else None
        ),
        routing_policy_id=clean.get("routing_policy_id", "skill_weighted"),
        priority=Priority(clean["priority"]),
        requested_completion=(
            datetime.fromisoformat(completion) if completion else None
        ),
        escalation_tier=clean.get("escalation_tier", 0),
    )

4. Apply preventive-maintenance routing precedence

Preventive maintenance requests carry temporal and conditional weight that reactive tickets do not. The validator preserves pm_trigger_type so the routing engine can apply correct precedence: calendar-based work lands in planned windows, while condition-based triggers (for example a vibration threshold breach) escalate to immediate skill-matched dispatch. The interval math that emits these triggers is handled upstream by PM interval calculation; the schema stage only translates the trigger into a priority and escalation tier.

def apply_pm_precedence(payload: WorkOrderPayload) -> WorkOrderPayload:
    """Promote condition-based PM work and pin escalation deterministically."""
    if payload.pm_trigger_type is None:
        return payload  # reactive work order — no PM promotion

    if payload.pm_trigger_type == PMTriggerType.CONDITION_BASED:
        promoted = Priority.CRITICAL
        tier = max(payload.escalation_tier, 2)
    elif payload.pm_trigger_type == PMTriggerType.CALENDAR:
        promoted = Priority.PLANNED
        tier = payload.escalation_tier
    else:  # runtime_hours / cycle_count — duty-driven, mid-priority
        promoted = Priority.STANDARD
        tier = payload.escalation_tier

    # Frozen dataclass: return a new instance rather than mutating.
    from dataclasses import replace
    return replace(payload, priority=promoted, escalation_tier=tier)

5. Publish to the routing engine

Validated payloads are serialized and posted to the CMMS over an authenticated endpoint. A monotonic work_order_id and an idempotency header let the API reject duplicates, so a network retry can never create two dispatches for one request.

import dataclasses
import logging
import requests

logger = logging.getLogger(__name__)


class WorkOrderPublisher:
    def __init__(self, cmms_base_url: str, api_token: str, timeout: int = 10):
        self.base_url = cmms_base_url.rstrip("/")
        self.session = requests.Session()
        self.session.headers.update({
            "Authorization": f"Bearer {api_token}",
            "Content-Type": "application/json",
            "X-Client-Id": "cmms-work-order-schema-v1",
        })
        self.timeout = timeout

    def publish(self, payload: WorkOrderPayload) -> bool:
        """Idempotent POST to the routing engine, keyed on work_order_id."""
        body = dataclasses.asdict(payload)
        # Enums and datetimes must serialize to primitive JSON.
        body["priority"] = payload.priority.value
        body["pm_trigger_type"] = (
            payload.pm_trigger_type.value if payload.pm_trigger_type else None
        )
        for ts in ("requested_completion", "created_at"):
            if body.get(ts) is not None:
                body[ts] = getattr(payload, ts).isoformat()

        try:
            resp = self.session.post(
                f"{self.base_url}/api/v1/work-orders",
                json=body,
                headers={"Idempotency-Key": payload.work_order_id},
                timeout=self.timeout,
            )
            resp.raise_for_status()
            logger.info("routed %s | priority:%s | zone:%s",
                        payload.work_order_id, payload.priority.value,
                        payload.location_zone)
            return True
        except requests.exceptions.HTTPError as e:
            if e.response is not None and e.response.status_code == 409:
                logger.warning("duplicate %s — skipping", payload.work_order_id)
                return False
            logger.error("HTTP error routing %s: %s", payload.work_order_id, e)
            raise
        except requests.exceptions.RequestException as e:
            logger.error("network/timeout routing %s: %s", payload.work_order_id, e)
            raise

Configuration Reference

Keep every tunable in a version-controlled configuration registry, not in the validator source. The defaults below are conservative starting points for a general-purpose maintenance routing service.

Validation and Testing

Schema verdicts must be reproducible, so the highest-value test asserts that a known-good request always constructs the same payload and a known-bad request always raises with the offending field named. A single deterministic assertion catches accidental drift — a relaxed pattern, a coercion regression, a silently widened enum — before it reaches production and starts admitting malformed work orders.

import pytest


def test_valid_request_constructs_payload():
    raw = {
        "work_order_id": "WO-8842",
        "asset_id": "f47ac10b-58cc-4372-a567-0e02b2c3d479",
        "required_skill_codes": ["HVAC-2"],
        "location_zone": "BLDG-204",
        "priority": "high",
        "escalation_tier": 1,
    }
    wo = normalize_and_validate(raw)
    assert wo.priority is Priority.HIGH
    assert wo.escalation_tier == 1
    assert wo.required_skill_codes == ["HVAC-2"]


def test_empty_skill_codes_is_rejected():
    raw = {
        "work_order_id": "WO-8843",
        "asset_id": "f47ac10b-58cc-4372-a567-0e02b2c3d479",
        "required_skill_codes": [],
        "location_zone": "BLDG-204",
        "priority": "standard",
    }
    with pytest.raises(SchemaRejection) as exc:
        normalize_and_validate(raw)
    assert exc.value.failed_field == "required_skill_codes"

On a successful publish, the routing engine receives a single structured log line per request — routed WO-8842 | priority:high | zone:BLDG-204 — which is the canonical signal that the work order was accepted. A 409 produces duplicate WO-8842 — skipping; seeing that line under retry storms confirms idempotency is holding rather than indicating a fault. Assert against both log lines in integration tests to verify the full validate-to-route path.

Failure Modes and Troubleshooting

Expand each scenario for the root cause, the diagnostic log excerpt, and the fix. The checklist items render as interactive checkboxes — work through them in order.

Valid-looking requests are rejected at the validation gate

• Diagnostic: SchemaRejection: pm_interval: '30d' is not of type 'integer' — a string slipped past normalization where the contract expects a number.
• Confirm the upstream adapter coerces meter values before submission; a '30d' interval means a unit suffix was never stripped.
• Root cause is almost always a type mismatch between the CMMS export format and the contract; coerce at the normalization boundary, never inside the validator.

Unknown fields silently break routing after an upstream change

• Check whether SCHEMA_STRICT_MODE was disabled — with additionalProperties allowed, a renamed field is carried through and the router reads a stale value.
• Diff the live payload against WORK_ORDER_SCHEMA["properties"]; any extra key is the culprit.
• Root cause is schema drift; re-enable strict mode and version the contract so adapter changes fail loudly at the seam.

Work orders route to the wrong crew or zone

• Verify location_zone matches the ^[A-Z]{2,}-[0-9]{3}$ pattern; a lowercase or mis-delimited code passes string typing but resolves to no zone.
• Confirm asset_id resolves against the asset master — an orphaned asset has no functional location for zone inference.
• Root cause is usually a zone-format mismatch or an unresolved asset; reconcile both against asset hierarchy design.

Rejected payloads vanish under load

• Look for the absence of any routed or duplicate line for affected requests — neither acceptance nor conflict logged.
• Confirm the API token still carries workorders:write; an expired or under-scoped token fails closed at publish.
• Check the dead-letter queue — rejections past max_retries should land there with error_code, failed_field, and payload_hash for replay rather than disappearing.

Frequently Asked Questions

Why keep the routing schema separate from the full work order record?

The routing schema is intentionally lean. It excludes compliance signatures, vendor SLAs, labor logs, and PII so dispatch stays fast and the attack surface stays small. The full record is assembled downstream once a technician is assigned; the schema stage only needs the fields required to make a deterministic routing decision.

Should validation run synchronously or in a worker?

Run it synchronously at the boundary so a malformed request is rejected before it consumes broker capacity or 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 that the worker pool provides.

How do SLA fields flow into routing decisions?

priority sets queue order, requested_completion arms the SLA timer, and escalation_tier defines how many hops occur before a supervisor is paged. Because these fields are mandatory on every payload, the routing engine never has to guess urgency — and the organization can prove response-time compliance from the same data it dispatched on.

What happens to a payload that fails validation?

It is raised as a structured SchemaRejection naming the offending field, then routed to a dead-letter queue with an error code and a payload hash. This converts what would otherwise be a silent drop into an explicit, reviewable rejection that integration teams can monitor and replay after fixing the upstream adapter.

Related

Anchor every asset_id to the equipment tree with asset hierarchy design, source PM triggers from PM interval calculation, gate dispatch access with security access boundaries, debug strict-mode edge cases in JSON schema validation for work order payloads, feed validated payloads through async batch processing, and confirm spares before dispatch via parts availability checks.

Part of: CMMS Architecture & Maintenance Taxonomy.