Automating Critical DVIR Defect Alerts to Dispatchers

The moment a driver certifies a severity 70–100 defect on a Driver Vehicle Inspection Report, the vehicle is out of service under 49 CFR § 396.11(a) and § 396.7 — and the single question this page answers is how to get that fact into the dispatcher’s hands, verifiably and exactly once, before the unit rolls. Get this wrong in either direction and the cost is concrete: a dropped alert leaves a legally out-of-service tractor available in the dispatch board, and dispatching it is a § 396.7(a) violation that surfaces in the Vehicle Maintenance BASIC at the next roadside inspection; a duplicated or noisy alert trains dispatchers to dismiss the interrupt, which is the same failure with a slower fuse. This how-to sits under the Critical vs Non-Critical Routing Logic engine and implements the notification leg of the critical branch that engine fires — it assumes the routing decision has already been made and concerns itself only with delivering it deterministically and idempotently.

Before implementing the alert router you need Python 3.10+ (for match statements and X | Y union syntax) and the following:

• pydantic>=2.6 — strict validation of the alert contract so a malformed payload is rejected, not delivered half-formed.
• httpx>=0.27 — connection-pooled HTTP with per-request timeouts for the dispatch webhook.
• tenacity>=8.2 — declarative exponential backoff with jitter, so retry policy is not hand-rolled.
• hashlib (stdlib) — SHA-256 idempotency keys for exactly-once delivery.
• A low-latency store (Redis or equivalent) for the idempotency key cache and the delivery-status ledger.

Two upstream contracts feed this page. The routing decision and severity band arrive from the Severity Scoring Algorithms for DVIR Defects layer, which emits the same 0–34 / 35–69 / 70–100 bands used across this site, and the canonical record you enrich the alert from is defined by the Standardized DVIR JSON Schema Design. This router receives an already-classified critical defect; it does not re-score.

Every field in the dispatcher alert traces to an operational or regulatory obligation. Keep this table as the authoritative source; the code below implements it field for field.

The single most-missed row is severity_band. This router must be unreachable for anything below 70; a 35–69 “major” defect belongs in the deferred maintenance queue, not on a dispatcher’s interrupt channel. Enforce the band at the router’s front door so a misrouted score fails closed rather than paging a dispatcher for a burned-out marker lamp.

Coerce the incoming payload into a strict Pydantic model so a missing or malformed field is rejected before any network call. The model also enforces the 70–100 band gate, so a sub-critical score can never reach the wire.

import hashlib
from datetime import datetime, timezone
from typing import Literal
from pydantic import BaseModel, Field, field_validator


class GeoPoint(BaseModel):
    lat: float
    lon: float


class DispatcherAlert(BaseModel):
    dvir_id: str
    unit_number: str
    driver_id: str                       # SHA-256 CDL hash, per 49 CFR 396.11(a)
    defect_code: str
    severity_band: int = Field(ge=70, le=100)   # critical band ONLY; fails closed below 70
    oos_status: Literal["OUT_OF_SERVICE"] = "OUT_OF_SERVICE"
    gps: GeoPoint | None = None
    shift_status: str | None = None
    cfr_basis: str = "49 CFR 396.7(a)"   # dispatching an OOS unit is a 396.7(a) violation
    detected_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))

    @field_validator("detected_at")
    @classmethod
    def _require_utc(cls, v: datetime) -> datetime:
        if v.tzinfo is None:
            raise ValueError("detected_at must be timezone-aware UTC")
        return v.astimezone(timezone.utc)


def build_alert(dvir_record: dict, routing_decision: dict) -> DispatcherAlert:
    """Merge the audit record with the routing decision into a validated alert."""
    return DispatcherAlert(
        dvir_id=dvir_record["dvir_id"],
        unit_number=dvir_record["unit_number"],
        driver_id=dvir_record["driver_cdl_hash"],
        defect_code=routing_decision["defect_code"],
        severity_band=routing_decision["severity_band"],
        gps=dvir_record.get("gps"),
        shift_status=dvir_record.get("shift_status"),
    )

Poor connectivity in the cab means the same critical defect can be submitted twice. Derive the idempotency key from the inspection identity — not from a timestamp or random value — so two submissions of the same defect collapse to one key and, therefore, one dispatcher interrupt.

def idempotency_key(alert: DispatcherAlert) -> str:
    """Stable across retries and duplicate submissions of the same defect."""
    material = f"{alert.dvir_id}:{alert.unit_number}:{alert.defect_code}"
    return hashlib.sha256(material.encode("utf-8")).hexdigest()

Check the key against a persistent ledger before spending a network call. A hit is a success no-op that returns the original delivery id — the dispatcher was already paged for this defect, and paging them again is the alert-fatigue failure mode.

import redis

_LEDGER = redis.Redis(decode_responses=True)
_LEDGER_TTL = 60 * 60 * 24 * 7   # 7 days covers any realistic retry window


def already_delivered(key: str) -> str | None:
    """Return the prior delivery id if this defect was already routed, else None."""
    return _LEDGER.get(f"alert:{key}")


def record_delivery(key: str, delivery_id: str) -> None:
    """Persist exactly-once proof; NX so a race cannot overwrite the first win."""
    _LEDGER.set(f"alert:{key}", delivery_id, nx=True, ex=_LEDGER_TTL)

Post the validated alert over a pooled httpx client with an explicit timeout, and retry only on transient failures. A 4xx (except 429) is a contract bug on our side — retrying it just delays the escalation, so it must fail fast.

import httpx
from tenacity import (
    retry, stop_after_attempt, wait_exponential_jitter,
    retry_if_exception_type,
)


class TransientDispatchError(Exception):
    """Raised for 5xx / 429 / network faults that are safe to retry."""


_CLIENT = httpx.Client(timeout=httpx.Timeout(5.0), http2=True)


@retry(
    retry=retry_if_exception_type(TransientDispatchError),
    wait=wait_exponential_jitter(initial=1, max=15),
    stop=stop_after_attempt(4),
    reraise=True,
)
def _post_alert(url: str, alert: DispatcherAlert, key: str, api_key: str) -> str:
    resp = _CLIENT.post(
        url,
        json=alert.model_dump(mode="json"),
        headers={
            "Authorization": f"Bearer {api_key}",
            "X-Idempotency-Key": key,   # webhook honors it too, for defense in depth
        },
    )
    if resp.status_code == 429 or 500 <= resp.status_code < 600:
        raise TransientDispatchError(f"transient {resp.status_code}")
    resp.raise_for_status()             # 4xx (non-429) fails fast, no retry
    return resp.json()["delivery_id"]

Tie the steps together. A critical alert that exhausts its retries must never be silently dropped — the unit is legally out of service, so escalate to an out-of-band channel and leave the vehicle held. Failing open here would put a § 396.7(a) violation on the road.

import logging

log = logging.getLogger("dispatch.alert_router")


def route_critical_alert(
    dvir_record: dict,
    routing_decision: dict,
    *,
    webhook_url: str,
    api_key: str,
    escalate,          # callable(alert) -> None : SMS/voice to safety manager
) -> str:
    alert = build_alert(dvir_record, routing_decision)   # Step 1 (validates band)
    key = idempotency_key(alert)                         # Step 2

    if (prior := already_delivered(key)) is not None:    # Step 3
        log.info("duplicate suppressed dvir=%s key=%s", alert.dvir_id, key)
        return prior

    try:
        delivery_id = _post_alert(webhook_url, alert, key, api_key)   # Step 4
    except (TransientDispatchError, httpx.HTTPError) as exc:
        log.critical("dispatch delivery failed dvir=%s: %s", alert.dvir_id, exc)
        escalate(alert)                                  # out-of-band fallback
        # Unit STAYS out of service; do not clear the hold on delivery failure.
        raise

    record_delivery(key, delivery_id)                    # Step 5
    log.info("alert delivered dvir=%s delivery=%s", alert.dvir_id, delivery_id)
    return delivery_id

Correctness here is about three invariants: the band gate holds, delivery is exactly-once, and a failed delivery never clears the hold. Assert each directly.

import pytest


def test_sub_critical_band_is_rejected():
    """A 35-69 'major' defect must never construct an alert."""
    with pytest.raises(ValueError):
        DispatcherAlert(
            dvir_id="d1", unit_number="4471", driver_id="h",
            defect_code="LAMP_MARKER", severity_band=52,
        )


def test_idempotency_key_is_stable_across_duplicates():
    a = DispatcherAlert(dvir_id="d1", unit_number="4471",
                        driver_id="h", defect_code="BRAKE_SYS", severity_band=88)
    b = a.model_copy(update={"detected_at": a.detected_at})  # different call, same identity
    assert idempotency_key(a) == idempotency_key(b)


def test_second_delivery_is_a_no_op(fake_ledger, fake_webhook):
    """The webhook must be hit once even when routing is invoked twice."""
    args = dict(webhook_url="https://x", api_key="k", escalate=lambda a: None)
    first = route_critical_alert(RECORD, DECISION, **args)
    second = route_critical_alert(RECORD, DECISION, **args)
    assert first == second
    assert fake_webhook.call_count == 1

Two assertions deserve emphasis. First, test_idempotency_key_is_stable_across_duplicates proves the key ignores volatile fields like detected_at — if a duplicate submission produced a fresh key, exactly-once delivery would silently break. Second, drive an integration test in which the webhook raises a 500 on all four attempts and assert that escalate fired and no delivery id was written to the ledger — that proves the hold survives a total delivery outage.

• Retrying a 4xx and burning the escalation window. A 422 from the webhook means the contract is wrong, not that the network is flaky. Blanket retry logic wraps it in backoff, delays the fallback by tens of seconds, and leaves an out-of-service unit unflagged that whole time. Retry only 429 and 5xx; fail fast on everything else and escalate.
• Timestamp-seeded idempotency keys. Keying on detected_at or a UUID means two submissions of the same cab-side defect produce two keys and two dispatcher pages. Derive the key from dvir_id + unit_number + defect_code only, and cache it before the POST so a mid-flight retry still collapses.
• Clearing the hold on delivery failure. The most dangerous bug is a finally block or optimistic status write that marks the DVIR “notified” even when every retry failed. The unit is out of service by regulation, not by whether dispatch got the message — the hold must persist until a human acknowledges the out-of-band escalation.
• Missing GPS blocking the alert. Enrichment fields like gps and shift_status are nice-to-have context, not gates. A validator that rejects the alert when GPS is None will suppress a legally required notification whenever a telematics unit is offline. Deliver the alert; attach location if you have it.

• Critical vs Non-Critical Routing Logic — the parent engine whose critical branch fires this alert.
• Severity Scoring Algorithms for DVIR Defects — the layer that emits the 70–100 band this router gates on.
• Building a Weighted Defect Scoring Model in Python — how the score feeding this router is computed and clipped.
• Mechanic Assignment & Workload Balancing — where the held unit routes once the mechanic picks up the repair.
• Standardized DVIR JSON Schema Design — the canonical record shape the alert is enriched from.

Part of the Critical vs Non-Critical Routing Logic guide. Back to Defect Classification & Repair Order Routing.