Escalating overdue calibration to lab managers

Problem statement #

You need a deterministic Python escalation engine that advances an out-of-calibration instrument through a fixed ladder — a due-soon reminder to the technician, an overdue notice, an escalation to the lab manager, and finally a compliance lockout — where each rung fires exactly once per instrument per window and every advance is recorded in an audit ledger, so an overdue analytical balance can never quietly keep charging a federal grant while its escalation history stays defensible under a site visit.

This task sits under Calibration Due Date Routing, one of the operational layers of Equipment Calibration & Lab Inventory Tracking. Routing decides which tier an instrument belongs to; escalation decides who is told, when, and how often once an instrument has already passed its due date. The two are complementary: the reminder cadence built in Automating Calibration Reminder Emails for Lab Equipment handles the pre-expiry nudges, while this page owns the post-expiry chain of accountability.

Prerequisites #

• Python 3.10+ (uses datetime.now(timezone.utc), union type hints, and hashlib.sha256).
• Libraries: pydantic>=2.5 for the escalation payload schema and SQLAlchemy>=2.0 with psycopg[binary] for the transactional ledger. Install with pip install "pydantic>=2.5" "SQLAlchemy>=2.0" "psycopg[binary]".
• Environment variables (never hard-code credentials, per Security Boundary Configuration):
  • CAL_DB_URL — the SQLAlchemy connection string for the escalation ledger.
• Policy config: a version-controlled table of days-overdue thresholds and their recipient sets, kept alongside your University Policy Mapping Frameworks. The instrument’s tier and due date are produced upstream by the routing engine; this page assumes a validated calibration record with a computed days_overdue is already in hand.

Step-by-step implementation #

The escalation ladder is a monotonic state machine. An instrument enters at DUE_SOON, and as days_overdue grows it can only advance — never regress on its own — through OVERDUE_TECH, ESCALATED_MANAGER, and finally LOCKED_OUT, at which point the instrument is blocked from grant-charged use and compliance is notified. Each advance emits a single notification and appends one immutable ledger row.

Figure: days-overdue advances an instrument up a monotonic ladder; each rung notifies one role and appends one ledger row.

Step 1 — Define the escalation tiers and the deterministic mapping #

The tier is a pure function of days_overdue and the instrument’s safety flag. Encoding the ladder as an ordered enum lets the engine compare rank — an instrument may only advance to a higher-ranked tier, which is what prevents flapping when a threshold sits near a day boundary.

import hashlib
import logging
from datetime import datetime, timezone
from enum import IntEnum
from typing import Any

logger = logging.getLogger("calibration_escalation")


class Tier(IntEnum):
    """Ordered escalation ladder; higher value = higher accountability."""
    DUE_SOON = 0          # technician reminder, pre/at expiry
    OVERDUE_TECH = 1      # technician owns remediation
    ESCALATED_MANAGER = 2 # lab manager is looped in
    LOCKED_OUT = 3        # compliance notified, instrument blocked


def classify_tier(days_overdue: int, safety_critical: bool) -> Tier:
    """Deterministic: same days_overdue + flag always yields the same tier.

    Safety-critical instruments escalate on a compressed schedule to satisfy
    the OSHA Laboratory Standard (29 CFR 1910.1450) tolerance for
    safety-relevant equipment.
    """
    if days_overdue <= 0:
        return Tier.DUE_SOON
    lockout_at = 14 if safety_critical else 30
    manager_at = 7 if safety_critical else 14
    if days_overdue > lockout_at:
        return Tier.LOCKED_OUT
    if days_overdue > manager_at:
        return Tier.ESCALATED_MANAGER
    return Tier.OVERDUE_TECH

Step 2 — Guard the transition so escalation only ever advances #

An instrument’s current tier lives in a persisted decision row. The transition function refuses any computed tier that is lower than the stored one: a clock adjustment, a re-imported certificate, or a late correction cannot walk an instrument back down the ladder and re-fire a lower notification. Only a verified recalibration — a genuinely new certificate producing a new identity — resets the chain.

def resolve_transition(stored_tier: Tier | None, computed_tier: Tier) -> Tier | None:
    """Return the tier to advance to, or None if no advance is warranted.

    Monotonic guard: never regress. This is what stops tier flapping when
    days_overdue oscillates around a threshold across successive runs.
    """
    if stored_tier is None:
        return computed_tier
    if computed_tier > stored_tier:
        return computed_tier
    return None  # same or lower rank -> no notification, no ledger row


def escalation_fingerprint(asset_id: str, cert_date: str, tier: Tier) -> str:
    """Stable SHA-256 over asset + certificate + tier.

    Binding the tier into the key means each rung has its own idempotency
    fingerprint, so a rung fires exactly once per instrument per certificate.
    """
    raw = f"{asset_id}|{cert_date}|{tier.name}"
    return hashlib.sha256(raw.encode("utf-8")).hexdigest()

Step 3 — Notify idempotently and append to the ledger in one transaction #

The driver computes the tier, resolves whether an advance is due, and — only when it is — sends the tier’s notification and appends a ledger row inside a single transaction. The notification send is keyed on the per-tier fingerprint, so a re-run of the same batch finds the fingerprint already recorded and sends nothing.

from sqlalchemy.dialects.postgresql import insert as pg_insert
from sqlalchemy.orm import Session


def escalate(
    payload: dict[str, Any],
    session: Session,
    notify,          # callable: (tier: Tier, recipients: list[str], ctx: dict) -> None
    recipients_for,  # callable: (tier: Tier, payload: dict) -> list[str]
    now: datetime | None = None,
) -> dict[str, Any]:
    """Advance one instrument's escalation exactly once per tier per window."""
    now = now or datetime.now(timezone.utc)
    due_date = payload["due_date"]                      # tz-aware UTC datetime
    days_overdue = (now - due_date).days
    computed = classify_tier(days_overdue, payload.get("safety_critical", False))

    stored = session.get(EscalationState, payload["asset_id"])
    stored_tier = Tier(stored.tier) if stored else None
    target = resolve_transition(stored_tier, computed)

    if target is None:
        logger.info("No advance | asset=%s tier=%s", payload["asset_id"], computed.name)
        return {"status": "no_op", "tier": computed.name}

    fp = escalation_fingerprint(payload["asset_id"], payload["cert_date"], target)

    # Idempotent ledger insert: ON CONFLICT DO NOTHING on the per-tier fingerprint
    # means a replay writes no second row and triggers no second notification.
    result = session.execute(
        pg_insert(EscalationLedger)
        .values(
            fingerprint=fp,
            asset_id=payload["asset_id"],
            tier=target.name,
            days_overdue=days_overdue,
            recorded_at=now,
        )
        .on_conflict_do_nothing(index_elements=["fingerprint"])
    )

    if result.rowcount:
        recipients = recipients_for(target, payload)
        notify(target, recipients, {"asset_id": payload["asset_id"], "days_overdue": days_overdue})
        session.merge(EscalationState(asset_id=payload["asset_id"], tier=target.value, updated_at=now))
        logger.info("Escalated | asset=%s -> %s", payload["asset_id"], target.name)

    session.commit()
    return {"status": "escalated", "tier": target.name, "fingerprint": fp}

Instruments that reach LOCKED_OUT are handed to the same quarantine path the routing layer uses to pause chargeback billing, and the preventive-maintenance work order they trigger is scheduled through Preventive Maintenance Scheduling.

Schema and field reference #

The escalation ledger is the compliance artifact. Keep the threshold-to-recipient mapping in version-controlled policy config, not in code.

Verification #

1. Deterministic classification: call classify_tier(days_overdue, flag) across a table of boundary values (0, 1, 7, 14, 30, 31) and confirm each returns the expected tier for both safety flags.
2. Monotonic guard: feed an instrument a decreasing days_overdue after it has reached ESCALATED_MANAGER; resolve_transition must return None and no ledger row may appear.
3. Idempotent replay: run escalate twice with the identical payload. Exactly one ledger row must exist for that tier and asset_id, and the notification hook must have been called once.
4. Reproduce the fingerprint: recompute escalation_fingerprint(asset_id, cert_date, tier) and confirm it equals SELECT fingerprint FROM escalation_ledger WHERE asset_id = :id AND tier = :tier.

Troubleshooting #

• Tier flapping re-sends notifications. When days_overdue oscillates around a threshold across nightly runs — often because the due date is timezone-naive and a run near local midnight computes a different day — an unguarded engine re-fires the lower tier. The monotonic resolve_transition guard is the fix: escalation is one-directional, so once an instrument is at ESCALATED_MANAGER a later run computing OVERDUE_TECH is a no-op. Confirm due dates are UTC-normalized before subtraction.
• A single tier double-sends across two workers. Two concurrent workers can both compute the same advance before either commits. The on_conflict_do_nothing on the unique per-tier fingerprint serializes them at the database: the first insert wins and sends, the second sees rowcount == 0 and sends nothing. Never send the notification before the successful insert.
• Recalibration does not reset the ladder. After a verified recalibration the instrument carries a new cert_date, which changes every per-tier fingerprint, so the chain starts fresh at DUE_SOON. If a re-calibrated instrument stays stuck at a high tier, confirm the new certificate date is flowing into the payload rather than the stale one.

Frequently asked questions #

Related #

• Up to the parent topic: Calibration Due Date Routing
• Automating Calibration Reminder Emails for Lab Equipment — the pre-expiry reminder cadence this ladder continues
• Preventive Maintenance Scheduling — the work orders a lockout triggers
• Equipment Calibration & Lab Inventory Tracking — the parent practice