Asset Hierarchy Design for CMMS Routing Pipelines

Asset hierarchy design is the routing-resolution stage of the CMMS Architecture & Maintenance Taxonomy domain — the component that turns a parent-child asset tree into the dispatch decision that sends a work order to the correct crew, zone, and escalation tier.

When the tree carries no routing metadata, the cost surfaces immediately: maintenance engineers receive jobs for equipment outside their trade, facilities managers watch labor pile up in a flat assignment queue while critical assets wait, and integration teams reconcile misrouted tickets by hand. Treating the hierarchy as a typed graph — where every node either declares its own routing attributes or inherits them from an ancestor — replaces that guesswork with a deterministic resolution that produces the same dispatch decision for the same trigger, every cycle. This guide implements that resolution 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

This component runs as a synchronous resolver invoked by the routing engine the moment a maintenance trigger is validated. Before you deploy it, confirm the following are in place.

• Python 3.11+ with networkx>=3.2 for the directed graph, pydantic>=2.6 for schema enforcement on node metadata, and requests>=2.31 for reading the asset registry over REST. No third-party scheduler is required; the resolver is driven inline by the validation stage upstream.
• CMMS REST API v1 with read access to the asset registry (GET /api/v1/assets/tree) and the routing rule resource (GET /api/v1/routing/rules). The API must return each node’s parent reference and its declared routing tags so the graph can be rebuilt deterministically on every sync.
• A resolved asset registry. Routing math is only meaningful once external equipment tags are mapped to canonical CMMS asset nodes. That mapping is anchored to the structure described here and consumed downstream by work order schema standards before a payload is dispatched.
• Environment variables: CMMS_BASE_URL, CMMS_API_TOKEN, and ROUTING_DEFAULT_ZONE (default FACILITY_QUEUE). The token must carry the assets:read and routing:read scopes; a token missing routing:read fails closed at graph load rather than silently resolving every job to the default queue.

Architecture and Data Contract

The component sits between trigger validation and work order dispatch. It never ingests raw maintenance signals directly — it consumes an already-validated trigger that references a canonical asset node, traverses the hierarchy to resolve routing attributes, and emits a typed routing decision. Four boundaries keep the resolution honest and stop downstream execution concerns from leaking into the tree:

• Ingestion boundary: a validated MaintenanceTrigger enters the resolver only after it has passed schema validation; no traversal runs against an unvalidated trigger.
• Resolution boundary: the deterministic upward traversal consumes a trigger plus the loaded graph and emits a RoutingDecision. Same tree, same trigger, same decision, every time.
• Validation boundary: the resolved decision is checked against the canonical work order schema before it leaves the component, so a malformed dispatch can never enter the execution queue.
• Dispatch boundary: the routing engine consumes the decision to assign labor, start SLA timers, and reserve parts through parts availability checks — none of which this component performs itself. The routing scope terminates the moment a valid dispatch_zone and primary_crew are resolved.

The contract across the resolution boundary is explicit. The input is a MaintenanceTrigger; the output is a RoutingDecision. Encoding the node metadata and both ends of the contract as pydantic models means a node missing a required tag, or a trigger that points at an unknown asset, is rejected at the boundary instead of producing a plausible-but-wrong dispatch.

from datetime import datetime
from enum import Enum
from typing import Optional
from pydantic import BaseModel, Field


class MaintenanceScope(str, Enum):
    CORRECTIVE = "corrective"
    PREVENTIVE = "preventive"
    PREDICTIVE = "predictive"


class MaintenanceTrigger(BaseModel):
    """Input contract: a validated trigger referencing one canonical asset node."""
    work_order_id: str = Field(..., min_length=6, max_length=64)
    trigger_asset_id: str = Field(..., min_length=4, max_length=64)
    scope: MaintenanceScope
    requested_completion: Optional[datetime] = None


class RoutingDecision(BaseModel):
    """Output contract: a fully resolved dispatch target for the routing engine."""
    work_order_id: str
    asset_id: str
    dispatch_zone: str
    primary_crew: str
    escalation_tier: int = Field(..., ge=0, le=5)
    scope: MaintenanceScope
    requested_completion: Optional[datetime] = None
    resolution_path: list[str]
    used_fallback: bool

The asset tree itself is modeled as a directed acyclic graph: edges point from parent to child, and a node’s routing attributes are validated when it is added so a malformed registry export is caught at load time rather than at dispatch time.

Step-by-Step Implementation

1. Define the canonical work order payload

Routing exists to stamp a dispatch target onto a work order, so the resolver imports the same WorkOrderPayload used everywhere on this site rather than redefining its own shape. The SLA fields — priority, requested_completion, and escalation_tier — are what the routing engine reads when it decides how urgently a resolved job must move and when it must escalate. Field-level schema details live in work order schema standards.

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"


@dataclass
class WorkOrderPayload:
    """Canonical CMMS work order — SLA fields are mandatory site-wide."""
    work_order_id: str
    asset_id: str
    part_skus: List[str]
    required_quantities: Dict[str, int]
    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. Validate node metadata and load the hierarchy into a DAG

Each node declares the routing tags it owns; tags it omits are inherited at resolution time. Modeling the tree as a networkx.DiGraph lets the loader reject circular dependencies up front — a cycle (A is the parent of B which is the parent of A) would make upward traversal loop forever, so it must fail at load, loudly, with the offending node identified.

import networkx as nx
from pydantic import BaseModel, Field
from typing import List, Optional, Tuple


class AssetNode(BaseModel):
    """Per-node routing metadata; unset tags inherit from an ancestor."""
    id: str = Field(..., min_length=4, max_length=64)
    parent_id: Optional[str] = None
    site_code: Optional[str] = None
    dispatch_zone: Optional[str] = None
    primary_crew: Optional[str] = None
    escalation_tier: Optional[int] = Field(default=None, ge=0, le=5)


class HierarchyGraph:
    ROUTING_TAGS = ("dispatch_zone", "primary_crew", "escalation_tier", "site_code")

    def __init__(self) -> None:
        self.graph = nx.DiGraph()

    def load(self, nodes: List[AssetNode]) -> None:
        """Build a parent->child DAG, validating metadata and rejecting cycles."""
        for node in nodes:
            attrs = {tag: getattr(node, tag) for tag in self.ROUTING_TAGS}
            self.graph.add_node(node.id, **attrs)
        for node in nodes:
            if node.parent_id is not None:
                if not self.graph.has_node(node.parent_id):
                    raise ValueError(
                        f"node {node.id!r} references missing parent {node.parent_id!r}"
                    )
                self.graph.add_edge(node.parent_id, node.id)
        if not nx.is_directed_acyclic_graph(self.graph):
            cycle = nx.find_cycle(self.graph)
            raise ValueError(f"asset hierarchy contains a cycle: {cycle}")

3. Resolve routing attributes by upward traversal with inheritance precedence

The resolver walks from the trigger asset toward the root, returning the first ancestor that declares the attribute it is looking for. Because explicit asset-level tags are encountered before site-level defaults, precedence falls out of the traversal order naturally: an asset override beats a system-group default, which beats a site default. Recording every node visited gives an auditable resolution_path for post-mortem analysis of any disputed dispatch.

from typing import Any, List, Optional


def resolve_attr(
    hierarchy: HierarchyGraph,
    asset_id: str,
    attr: str,
) -> Tuple[Optional[Any], List[str]]:
    """Walk parent->...->root; return the nearest defined value and the path taken."""
    graph = hierarchy.graph
    if not graph.has_node(asset_id):
        return None, []

    path: List[str] = []
    current: Optional[str] = asset_id
    visited: set[str] = set()

    while current is not None and current not in visited:
        visited.add(current)
        path.append(current)
        value = graph.nodes[current].get(attr)
        if value is not None:
            return value, path
        # A DAG node has at most one structural parent in an asset tree.
        parents = list(graph.predecessors(current))
        current = parents[0] if parents else None

    return None, path

4. Build a validated routing decision with deterministic fallback

When upward traversal exhausts the tree without finding a primary_crew or dispatch_zone, the resolver must not raise — it falls back to a facility-level queue and flags the decision so operations can backfill the missing metadata. The decision is validated against its pydantic contract before it is returned, so an out-of-range escalation tier or an empty crew can never reach the dispatch engine.

import logging

logger = logging.getLogger(__name__)


class RoutingResolver:
    def __init__(self, hierarchy: HierarchyGraph, default_zone: str = "FACILITY_QUEUE",
                 default_crew: str = "GENERAL_MAINT") -> None:
        self.hierarchy = hierarchy
        self.default_zone = default_zone
        self.default_crew = default_crew

    def resolve(self, trigger: MaintenanceTrigger) -> RoutingDecision:
        """Resolve a validated trigger into a dispatch decision with fallback."""
        zone, zone_path = resolve_attr(self.hierarchy, trigger.trigger_asset_id, "dispatch_zone")
        crew, _ = resolve_attr(self.hierarchy, trigger.trigger_asset_id, "primary_crew")
        tier, _ = resolve_attr(self.hierarchy, trigger.trigger_asset_id, "escalation_tier")

        used_fallback = zone is None or crew is None
        if used_fallback:
            logger.warning(
                "routing fallback for %s | asset:%s — missing zone/crew",
                trigger.work_order_id, trigger.trigger_asset_id,
            )

        decision = RoutingDecision(
            work_order_id=trigger.work_order_id,
            asset_id=trigger.trigger_asset_id,
            dispatch_zone=zone or self.default_zone,
            primary_crew=crew or self.default_crew,
            escalation_tier=tier if tier is not None else 0,
            scope=trigger.scope,
            requested_completion=trigger.requested_completion,
            resolution_path=zone_path,
            used_fallback=used_fallback,
        )
        logger.info(
            "routed %s | asset:%s -> zone:%s crew:%s tier:%s",
            decision.work_order_id, decision.asset_id,
            decision.dispatch_zone, decision.primary_crew, decision.escalation_tier,
        )
        return decision

The traversal can be visualized as a climb from a leaf asset up through its system group and site to the first ancestor that owns each tag.

Configuration Reference

Keep every routing tunable in a version-controlled configuration registry, not in the resolver source. The defaults below are conservative starting points for a multi-building facility.

Validation and Testing

Routing must be reproducible, so the highest-value test asserts that the same tree and trigger always yield the same decision, and that inheritance precedence holds. A single deterministic assertion catches accidental nondeterminism — an unstable parent ordering, a clock-dependent default — before it reaches production.

def test_resolution_is_deterministic_and_inherits():
    nodes = [
        AssetNode(id="SITE-EAST", dispatch_zone="ZONE-E", site_code="EAST"),
        AssetNode(id="CHILLER-PLANT", parent_id="SITE-EAST", primary_crew="HVAC-CREW"),
        AssetNode(id="PMP-204", parent_id="CHILLER-PLANT"),
    ]
    hierarchy = HierarchyGraph()
    hierarchy.load(nodes)
    resolver = RoutingResolver(hierarchy)

    trigger = MaintenanceTrigger(
        work_order_id="WO-100042",
        trigger_asset_id="PMP-204",
        scope=MaintenanceScope.PREVENTIVE,
    )
    a = resolver.resolve(trigger)
    b = resolver.resolve(trigger)
    assert a == b
    # PMP-204 inherits crew from CHILLER-PLANT and zone from SITE-EAST.
    assert a.primary_crew == "HVAC-CREW"
    assert a.dispatch_zone == "ZONE-E"
    assert a.used_fallback is False
    assert a.resolution_path == ["PMP-204", "CHILLER-PLANT", "SITE-EAST"]

On a successful resolution the resolver emits a single structured log line per work order — routed WO-100042 | asset:PMP-204 -> zone:ZONE-E crew:HVAC-CREW tier:0 — which is the canonical signal that a dispatch target was found. A fallback produces routing fallback ... missing zone/crew; seeing that line spike after a registry import points at nodes that lost their tags during the export. Assert against both log lines in integration tests to verify the full trigger-to-decision 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.

Every job resolves to FACILITY_QUEUE

• Confirm the API token still carries routing:read; an under-scoped token loads the tree without routing tags and every node falls back.
• Look for a burst of routing fallback ... missing zone/crew lines immediately after a registry sync — that timing points at a lossy export, not a code bug.
• Root cause is almost always tags stripped during the GET /api/v1/assets/tree export; reconcile the exported node count and tag coverage against the source registry.

Resolver hangs or raises on load

• Check the load step for asset hierarchy contains a cycle: [...]; the listed edges name the parent-child loop to break.
• Verify a node’s parent_id was not accidentally set to its own id during a bulk edit, which creates a self-loop.
• If the registry legitimately contains a malformed subtree, switch cycle_policy to quarantine so the rest of the tree still loads while the bad subtree is dead-lettered for repair.

Jobs routed to a crew at the wrong facility

• Diagnostic: a resolution_path that climbs past the expected site root into another facility’s subtree.
• Confirm site_isolation is true so traversal stops at the site boundary; cross-facility dispatch bleed almost always means it was disabled.
• Verify parent-child edges reflect physical and logical dependencies — a node mis-parented under another site’s group will inherit that site’s crew. Re-anchor it using how to structure CMMS asset trees for multi-site facilities.

A specialized job lands with the general crew despite a correct tag

• Confirm inheritance_mode is nearest_ancestor; in strict_explicit an asset with no own tag falls straight to the default crew.
• Check precedence — an explicit primary_crew on the asset must short-circuit the climb; if it does not, the tag is set on the wrong node level.
• Validate the resolved decision against work order schema standards; a tier or crew that fails the contract is dead-lettered rather than dispatched.

Frequently Asked Questions

Should the asset graph be rebuilt on every trigger or cached?

Rebuild it from the registry on a sync cadence and cache the in-memory DiGraph between triggers; resolution itself is a cheap read-only traversal. Invalidate the cache on any registry write so a re-parented node or a changed tag takes effect on the next trigger rather than after a restart.

How do I keep one facility from inheriting another facility’s crews?

Keep site_isolation enabled so upward traversal stops at the site boundary, and make sure every site root declares its own dispatch_zone and default crew. Cross-facility dispatch bleed is almost always a mis-parented node or a disabled isolation flag, not a flaw in the traversal.

What happens when a trigger references an asset that is not in the tree?

resolve_attr returns an empty path, the resolver flags used_fallback, and the job goes to FACILITY_QUEUE for human triage rather than being dropped. Treat a rising fallback rate as a signal that the asset registry and the trigger source have drifted out of sync.

Where should routing tags live — on assets, system groups, or sites?

Declare each tag at the highest level where it is uniformly true and override only where reality differs. Put dispatch_zone at the site or building level, primary_crew at the system-group level, and reserve asset-level tags for genuine exceptions. Inheritance then keeps the registry small and the precedence rules predictable.

Related

Validate every resolved decision against work order schema standards, set the cadence that fires preventive triggers with PM interval calculation, constrain who can edit the tree and its tags through security access boundaries, and design topology for distributed estates with how to structure CMMS asset trees for multi-site facilities.

Part of: CMMS Architecture & Maintenance Taxonomy.