SLA Timers and Escalation Enforcement in CMMS Dispatch

SLA timers and escalation enforcement are the deadline-keeping stage of the Technician Routing & SLA Enforcement pipeline — the component that converts a work order’s priority into a hard due time, tracks the elapsed clock across every state transition, and fires tiered escalations before the deadline is missed rather than after.

Routing a technician to the right job with the right parts still fails the business if the job finishes late. This component enforces the promise: it derives a due_at timestamp the moment a work order is accepted, accrues elapsed time only while the order is genuinely being worked, pauses the clock for legitimate holds, and escalates through a tier ladder as remaining time thins. For facilities managers and reliability engineers it turns an abstract service-level agreement into an auditable countdown with predictable warning points; for Python automation and CMMS integration teams it demands correct business-hours math, timezone-safe timestamps, idempotent escalation firing, and a state machine that cannot leak time or double-notify. This guide builds that stage end to end — prerequisites, the data contract and state machine, a step-by-step implementation, a configuration matrix, validation checks, and the failure modes you will actually hit in production.

Prerequisites

The timer runs as a scheduled evaluator: a periodic job wakes, recomputes remaining time for every open clock, and emits escalation events. Before you deploy it, confirm the following are in place.

• Python 3.11+ with pydantic>=2.6 for the clock model, python-dateutil>=2.9 for DST-safe arithmetic and recurrence, and a scheduler such as APScheduler>=3.10 (or a Celery beat driven by the broker from async batch processing) to run the evaluation tick every 30–60 seconds.
• A message broker or task queue so escalation notifications fire out-of-band from the tick loop. The evaluator decides that an escalation is due; a worker delivers it. Coupling delivery into the tick makes one slow webhook stall every other clock.
• A business-hours calendar. SLA math is meaningless in wall-clock seconds when the agreement is stated in business hours. Source shift windows, holidays, and coverage gaps from shift calendar integration so the clock only advances while the site is actually staffed.
• Persisted clock state. Each work order owns one durable SLAClock row (due time, tier, accumulated pause). The tick is stateless; all truth lives in the store so a restart never loses or double-counts elapsed time.
• Environment variables: SLA_TICK_INTERVAL_S (default 45), SLA_DEFAULT_TZ (an IANA name such as America/Chicago), ESCALATION_WEBHOOK_URL, and SLA_MATRIX_PATH pointing at the version-controlled matrix. The escalation identity carrying the sla:escalate scope must be present or the notification step fails closed and logs, rather than silently dropping a breach warning.
• Permissions. The tick reads work orders and writes only to the clock and the escalation log; it never mutates the work order’s own fields. Escalation records are written through compliance audit logging so every tier change is attributable.

Architecture and Data Contract

The enforcer sits alongside the dispatch flow, not inside it. Dispatch decides who works a job; the enforcer decides when it is late and who to wake as that moment approaches. It reads the canonical work order, owns a single clock per order, and emits escalation events — it never reassigns technicians and never closes work orders. Four boundaries keep the countdown honest.

• Derivation boundary: when a work order is accepted, its priority and escalation_tier resolve — once — to a due_at through the SLA matrix and the business-hours calendar. The due time is stored, not recomputed on every read, so a matrix edit never retroactively rewrites a live commitment.
• Accrual boundary: elapsed time advances only in states that count against the SLA. Entering a paused state (waiting on parts, waiting on access) suspends accrual; resuming closes the pause interval. Remaining time is due_at minus business-hours elapsed, never wall-clock elapsed.
• Escalation boundary: each tick compares remaining time against tier thresholds and, when a threshold is crossed, bumps the tier and enqueues exactly one notification. Firing is idempotent — the same threshold cannot notify twice.
• Breach boundary: when remaining time reaches zero the clock is marked breached, a terminal escalation fires, and the event is written to the audit trail. A resolved order stops its clock; a breached-then-resolved order stays flagged for reporting.

The state machine below is the contract for how a clock moves. Accrual runs in assigned and in_progress; it suspends in paused; it stops in resolved. Escalation checkpoints are evaluated on every tick against whichever state is active.

The contract has two halves. The input is the canonical work order — the enforcer reads priority and escalation_tier, whose meaning and validation are fixed in work order schema standards. The output is an SLAClock: a durable record of the derived due time, the current tier, and the closed and open pause intervals that let remaining time be recomputed deterministically on any tick.

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]
    location_id: str
    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))

The clock model carries only what the enforcer owns. Pauses are stored as (start, end) intervals; an open pause has end=None. Everything else is derived, so a restarted tick reconstructs remaining time from the store alone.

from pydantic import BaseModel, Field
from datetime import datetime
from typing import List, Optional, Tuple


class SLAClock(BaseModel):
    """Durable SLA state for exactly one work order."""
    work_order_id: str = Field(..., min_length=8, max_length=64)
    priority: Priority
    tier: int = Field(0, ge=0, le=3)
    started_at: datetime                       # when the clock began accruing
    due_at: datetime                            # derived business-hours deadline
    pauses: List[Tuple[datetime, Optional[datetime]]] = Field(default_factory=list)
    last_fired_tier: int = Field(0, ge=0, le=3)  # idempotency guard for escalations
    breached: bool = False
    resolved_at: Optional[datetime] = None

Step-by-Step Implementation

1. Derive the due time from priority and escalation tier

When a work order is accepted, resolve its priority and escalation_tier to a response budget through the SLA matrix, then project that budget forward across the business-hours calendar to land a real due_at. Deriving once and persisting the result is what keeps a live commitment stable — a later matrix change must not silently move an existing deadline. Adding business hours is not adding wall-clock hours: the projection walks the shift calendar and skips closed windows.

from datetime import datetime, timedelta
from typing import Callable

# SLA budgets in business hours, keyed by (priority, escalation_tier).
SLA_MATRIX = {
    (Priority.CRITICAL, 0): 4,
    (Priority.CRITICAL, 1): 2,
    (Priority.HIGH, 0): 8,
    (Priority.HIGH, 1): 4,
    (Priority.STANDARD, 0): 24,
    (Priority.STANDARD, 1): 16,
    (Priority.PLANNED, 0): 72,
}


def resolve_budget_hours(priority: Priority, escalation_tier: int) -> int:
    """Look up the response budget, falling back to the tier-0 row."""
    if (priority, escalation_tier) in SLA_MATRIX:
        return SLA_MATRIX[(priority, escalation_tier)]
    return SLA_MATRIX[(priority, 0)]


def derive_due_at(
    started_at: datetime,
    priority: Priority,
    escalation_tier: int,
    add_business_hours: Callable[[datetime, float], datetime],
) -> datetime:
    """Project the budget across the shift calendar to a concrete deadline."""
    budget = resolve_budget_hours(priority, escalation_tier)
    # add_business_hours is supplied by the shift-calendar integration; it walks
    # open shift windows and skips holidays/closed hours.
    return add_business_hours(started_at, budget)

2. Account for elapsed and remaining time across pauses

Remaining time is the budget minus the business-hours actually consumed, and consumed time excludes every paused interval. Compute business-hours elapsed between the start and now, subtract the business-hours inside each closed pause, and subtract an open pause up to now. Keeping the math pure — a function of stored timestamps only — is what makes it reproducible on any tick and testable without a clock.

def paused_business_hours(
    clock: SLAClock,
    now: datetime,
    business_hours_between: Callable[[datetime, datetime], float],
) -> float:
    """Total business hours spent in pauses, including any open interval."""
    total = 0.0
    for start, end in clock.pauses:
        total += business_hours_between(start, end or now)
    return total


def remaining_hours(
    clock: SLAClock,
    now: datetime,
    business_hours_between: Callable[[datetime, datetime], float],
) -> float:
    """Budget minus consumed business hours; consumed excludes pauses."""
    gross = business_hours_between(clock.started_at, clock.due_at)  # the budget
    elapsed = business_hours_between(clock.started_at, now)
    paused = paused_business_hours(clock, now, business_hours_between)
    consumed = elapsed - paused
    return gross - consumed


def remaining_fraction(clock: SLAClock, now: datetime,
                       business_hours_between: Callable[[datetime, datetime], float]) -> float:
    """Remaining time as a fraction of the total budget, clamped to [0, 1]."""
    gross = business_hours_between(clock.started_at, clock.due_at)
    if gross <= 0:
        return 0.0
    return max(0.0, min(1.0, remaining_hours(clock, now, business_hours_between) / gross))

A pause is opened when the work order enters a hold — most commonly waiting on parts — and closed when work resumes. The rule that decides whether a hold should pause the clock is deliberately narrow; the detailed policy for parts holds lives in pausing SLA clocks for parts-on-order holds, which draws its signal from the parts availability checks gate.

def open_pause(clock: SLAClock, at: datetime) -> SLAClock:
    """Begin a pause; a no-op if one is already open (idempotent)."""
    if clock.pauses and clock.pauses[-1][1] is None:
        return clock
    clock.pauses.append((at, None))
    return clock


def close_pause(clock: SLAClock, at: datetime) -> SLAClock:
    """Close the open pause; a no-op if none is open."""
    if clock.pauses and clock.pauses[-1][1] is None:
        start, _ = clock.pauses[-1]
        clock.pauses[-1] = (start, at)
    return clock

3. Bump the escalation tier when a threshold is crossed

On each tick, translate remaining fraction into the tier it warrants, then bump only when the warranted tier exceeds the tier already fired. The last_fired_tier guard is the idempotency mechanism: it makes escalation a monotonic ratchet so a clock hovering at 49% cannot re-notify the lead on every tick. The function returns the new tier and whether a notification is owed, keeping the decision separable from delivery.

from typing import Tuple

# Escalation ladder: fraction remaining at or below which each tier is warranted.
TIER_THRESHOLDS = [
    (1, 0.50),  # tier 1 at 50% remaining -> notify lead
    (2, 0.20),  # tier 2 at 20% remaining -> notify supervisor
    (3, 0.00),  # tier 3 at breach        -> terminal
]


def warranted_tier(fraction: float) -> int:
    """Highest tier whose threshold the remaining fraction has crossed."""
    tier = 0
    for level, threshold in TIER_THRESHOLDS:
        if fraction <= threshold:
            tier = level
    return tier


def evaluate_escalation(clock: SLAClock, fraction: float) -> Tuple[SLAClock, bool]:
    """Ratchet the tier upward; return (clock, notification_owed)."""
    target = warranted_tier(fraction)
    if target > clock.last_fired_tier:
        clock.tier = target
        clock.last_fired_tier = target
        return clock, True
    return clock, False

4. Detect a breach and fire the notification hook

Compose the per-clock tick: skip resolved clocks, recompute remaining time, mark a breach when it hits zero, run the escalation ratchet, and enqueue exactly one notification when one is owed. Delivery is a hook, not an inline call — the tick enqueues an event and returns, so a slow webhook never stalls the evaluation of the next clock. Every tier change is written to the audit trail through compliance audit logging so escalations are attributable after the fact.

import logging
from datetime import datetime, timezone
from typing import Callable, Optional

logger = logging.getLogger(__name__)


def tick_clock(
    clock: SLAClock,
    now: datetime,
    business_hours_between: Callable[[datetime, datetime], float],
    enqueue_notification: Callable[[str, int, float], None],
    record_audit: Callable[[str, int, bool], None],
) -> SLAClock:
    """Advance one clock: recompute, ratchet, and fire at most one escalation."""
    if clock.resolved_at is not None:
        return clock

    fraction = remaining_fraction(clock, now, business_hours_between)

    if fraction <= 0.0 and not clock.breached:
        clock.breached = True
        logger.warning("sla breach wo:%s tier:%s", clock.work_order_id, clock.tier)

    clock, owed = evaluate_escalation(clock, fraction)
    if owed:
        # Deliver out-of-band; the tick only enqueues.
        enqueue_notification(clock.work_order_id, clock.tier, fraction)
        record_audit(clock.work_order_id, clock.tier, clock.breached)
        logger.info(
            "sla escalation wo:%s tier:%s remaining:%.2f breached:%s",
            clock.work_order_id, clock.tier, fraction, clock.breached,
        )
    return clock


def run_tick(
    clocks: list[SLAClock],
    business_hours_between: Callable[[datetime, datetime], float],
    enqueue_notification: Callable[[str, int, float], None],
    record_audit: Callable[[str, int, bool], None],
    now: Optional[datetime] = None,
) -> list[SLAClock]:
    """Scheduler entry point: evaluate every open clock once."""
    now = now or datetime.now(timezone.utc)
    return [
        tick_clock(c, now, business_hours_between, enqueue_notification, record_audit)
        for c in clocks
    ]

5. Resolve the clock and stop accrual

When the work order reaches resolved, close any open pause and stamp resolved_at so subsequent ticks skip it. A clock that breached before resolution keeps its breached flag for reporting — resolving late does not erase the miss, it only stops the countdown.

def resolve_clock(clock: SLAClock, at: datetime) -> SLAClock:
    """Stop the clock; preserve the breach flag for reporting."""
    clock = close_pause(clock, at)
    clock.resolved_at = at
    logger.info(
        "sla resolved wo:%s breached:%s final_tier:%s",
        clock.work_order_id, clock.breached, clock.tier,
    )
    return clock

Configuration Reference

Keep the SLA matrix and every tunable in a version-controlled registry loaded from SLA_MATRIX_PATH, never inline in the tick source — a deadline policy is a governed artifact, not a code constant. The defaults below are conservative starting points for general-purpose MRO operations on a single-timezone site.

Validation and Testing

The accrual and escalation math is pure — every function takes explicit timestamps and calendar callbacks — so the highest-value tests inject a deterministic calendar and assert exact outcomes. A wall-clock-free test proves that a pause returns time to the budget and that the ratchet fires each tier exactly once.

from datetime import datetime, timedelta, timezone


def flat_business_hours_between(a: datetime, b: datetime) -> float:
    """Test calendar: treat every hour as a business hour."""
    return (b - a).total_seconds() / 3600.0


def test_pause_returns_time_to_the_budget():
    start = datetime(2026, 7, 16, 8, 0, tzinfo=timezone.utc)
    clock = SLAClock(
        work_order_id="WO-2026-0500",
        priority=Priority.HIGH,
        started_at=start,
        due_at=start + timedelta(hours=8),   # 8h budget
    )
    now = start + timedelta(hours=6)
    # Without a pause, 6h consumed -> 2h (25%) remaining.
    assert abs(remaining_hours(clock, now, flat_business_hours_between) - 2.0) < 1e-6
    # Pause 3 of those hours -> only 3h consumed -> 5h (62.5%) remaining.
    clock.pauses.append((start + timedelta(hours=1), start + timedelta(hours=4)))
    assert abs(remaining_hours(clock, now, flat_business_hours_between) - 5.0) < 1e-6


def test_escalation_ratchet_fires_each_tier_once():
    clock = SLAClock(
        work_order_id="WO-2026-0501",
        priority=Priority.CRITICAL,
        started_at=datetime(2026, 7, 16, 8, 0, tzinfo=timezone.utc),
        due_at=datetime(2026, 7, 16, 12, 0, tzinfo=timezone.utc),
    )
    clock, owed_a = evaluate_escalation(clock, 0.45)   # crosses tier 1
    clock, owed_b = evaluate_escalation(clock, 0.42)   # still tier 1, no re-fire
    clock, owed_c = evaluate_escalation(clock, 0.10)   # crosses tier 2
    assert (owed_a, owed_b, owed_c) == (True, False, True)
    assert clock.last_fired_tier == 2

On each escalation the tick emits one structured line — sla escalation wo:WO-2026-0501 tier:2 remaining:0.10 breached:False — which is the canonical signal that a tier changed against live remaining time. A breach adds sla breach wo:... tier:... immediately before the escalation line; seeing that pair confirms the terminal path fired exactly once. Assert against both lines in integration tests to verify the full derive-tick-escalate path, and confirm no duplicate escalation line shares a work_order_id and tier.

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.

The clock keeps accruing while a work order waits on parts

• Confirm the hold transition actually calls open_pause — a status change to a waiting state that does not open a pause interval leaves accrual running.
• Check that the parts-hold signal is wired from the parts availability checks gate; a DEFER_PROCUREMENT directive that never reaches the enforcer means no pause is ever opened.
• Verify the pause is closed on resume — an interval left open with end=None is correct while paused but must close when work restarts, or the clock never resumes accruing.

Escalations fire an hour early or late after a clock change

• Confirm every stored timestamp is timezone-aware and anchored to SLA_DEFAULT_TZ, not a naive server-local value; a naive datetime silently shifts across a DST boundary.
• Check that business-hours math runs through the calendar from shift calendar integration rather than raw timedelta subtraction — wall-clock arithmetic ignores the spring-forward and fall-back hour.
• Reproduce with a due_at that spans a DST transition; if remaining time jumps by an hour, the calendar callback is doing naive subtraction and must use dateutil zone-aware arithmetic.

An escalation storm notifies the same lead dozens of times

• Look for repeated sla escalation wo:<id> tier:<n> lines sharing the same work_order_id and tier — their presence means the ratchet is not holding.
• Confirm last_fired_tier is persisted with the clock and reloaded on each tick; if the tick reads a fresh in-memory clock, the guard resets and re-fires every cycle.
• Verify delivery honors escalation_dedupe_window_s so a double-write under a store race collapses to one send rather than two notifications.

A restart double-counts elapsed time or loses a pause

• Diagnostic: remaining time drops sharply right after a deploy for clocks that were paused across the restart.
• Confirm the tick is stateless and reconstructs remaining time from stored started_at, due_at, and pauses only — any elapsed counter held in process memory is lost on restart.
• Ensure an open pause interval is flushed to the store before shutdown so the end=None marker survives; an in-memory-only open pause is silently dropped and its time wrongly counted against the SLA.

Frequently Asked Questions

How is the SLA due time computed from priority and tier?

The enforcer looks up a response budget in the SLA matrix keyed on priority and escalation_tier, then projects that budget across the business-hours calendar to land a concrete due_at. The projection walks open shift windows and skips closed hours and holidays, so an eight-hour budget started late Friday lands during Monday’s shift rather than over the weekend. The due time is derived once at acceptance and persisted, so a later matrix edit never rewrites a live commitment.

Why track pauses as intervals instead of subtracting a running total?

Because a restart must not lose accumulated pause time and a tick must be reproducible from stored state alone. Storing each pause as a (start, end) interval lets any tick recompute consumed time deterministically from timestamps, with no in-process counter to lose. It also yields an auditable record of exactly when and how long the clock was suspended.

What stops a clock near a threshold from escalating on every tick?

The last_fired_tier guard makes escalation a monotonic ratchet. A tier only fires when the warranted tier exceeds the tier already fired, so a clock hovering at 49% remaining crosses tier 1 once and stays quiet until it crosses tier 2. A delivery-side dedupe window catches the rarer case of the tier being written twice under a store race.

Does resolving a work order late clear the breach?

No. Resolving closes any open pause and stops the countdown, but a clock that reached zero remaining keeps its breached flag for reporting. Late resolution stops further escalation; it does not erase the miss, which is what compliance reporting depends on.

Related

Draw the response budgets from the fields fixed in work order schema standards, anchor the countdown to real coverage with shift calendar integration, suspend the clock correctly with pausing SLA clocks for parts-on-order holds, record every tier change through compliance audit logging, and make sure the right technician is already on the job with skill-based dispatch.

Part of: Technician Routing & SLA Enforcement.