Routing Compliance Alerts with Severity Tiers

The exact task this guide solves is turning a stream of duty-time verdicts into tiered, routed alerts — an informational note logged quietly, a warning surfaced to the scheduler, an escalation paged to a compliance officer — without coupling the routing to the validation path. An engine that treats every finding identically either spams the escalation channel or buries a real violation in an activity log. The remedy is a severity classifier feeding an asynchronous router, so a hard violation reaches a human immediately while a near-minimum note is recorded for review, and neither blocks the scheduling UI. This page builds the tiering and the non-blocking dispatch. It completes the threshold tuning and alerting section and consumes the tuned levels from reducing alarm fatigue.

Prerequisites

Python 3.11+ with asyncio for non-blocking dispatch.
A verdict stream — each carrying an alert level, the rule violated, and the affected pairing.
A routing table — mapping severity tiers to destinations (log, scheduler queue, on-call escalation).
A message broker or queue so dispatch never blocks the scheduling path.

The three tiers and their destinations

Three tiers cover the operational need. Informational covers a duty that is legal but close to a limit or carries elevated fatigue — logged for trend review, no one paged. Warning covers a duty that breaches a contractual buffer or a soft policy — routed to the scheduler who owns the pairing, to be resolved before publication. Escalation covers a hard regulatory violation or a repeated warning ignored past a deadline — paged to a compliance officer with the full context. The classifier’s job is to map each verdict to its tier; the router’s job is to deliver it to the right place without stalling validation.

Step 1 — Classify a verdict into a severity tier

The tier derives from the alert level and whether the breach is of a regulatory limit or a softer buffer. Encoding the mapping as data keeps the escalation policy auditable.

from dataclasses import dataclass
from enum import Enum


class Tier(Enum):
    INFO = "informational"
    WARNING = "warning"
    ESCALATION = "escalation"


@dataclass(frozen=True)
class Verdict:
    pairing_id: str
    rule: str
    level: str          # "ok" | "warn" | "violation"
    regulatory: bool    # True if a hard regulatory limit, False if a buffer/policy


def severity(v: Verdict) -> Tier:
    if v.level == "violation" and v.regulatory:
        return Tier.ESCALATION
    if v.level == "violation" or v.level == "warn":
        return Tier.WARNING if not v.regulatory else Tier.ESCALATION
    return Tier.INFO

Verify: a regulatory violation classifies as ESCALATION; a contractual-buffer warn classifies as WARNING; a near-minimum ok-with-context classifies as INFO.

Step 2 — Route each tier to its destination

A routing table maps tiers to handlers, so adding a destination or re-pointing a tier is a configuration change, not a code edit. Each handler is an async callable that publishes to its channel.

async def to_audit_log(v: Verdict) -> None: ...
async def to_scheduler_queue(v: Verdict) -> None: ...
async def to_oncall(v: Verdict) -> None: ...


ROUTES = {
    Tier.INFO: [to_audit_log],
    Tier.WARNING: [to_audit_log, to_scheduler_queue],
    Tier.ESCALATION: [to_audit_log, to_scheduler_queue, to_oncall],
}


async def route(v: Verdict) -> None:
    for handler in ROUTES[severity(v)]:
        await handler(v)

Verify: an escalation fans out to all three handlers — it is always logged and queued as well as paged — while an informational verdict reaches only the audit log, so no one is paged for a legal near-miss.

Step 3 — Dispatch asynchronously so validation never blocks

Routing must not stall the validation sweep or the scheduling UI. Publishing verdicts to a bounded async queue and draining it with a worker pool decouples the two: a slow escalation channel cannot back-pressure the engine.

import asyncio


async def dispatch_all(verdicts: list[Verdict], *, workers: int = 8) -> None:
    queue: asyncio.Queue = asyncio.Queue()
    for v in verdicts:
        queue.put_nowait(v)

    async def worker() -> None:
        while not queue.empty():
            await route(await queue.get())
            queue.task_done()

    async with asyncio.TaskGroup() as tg:
        for _ in range(workers):
            tg.create_task(worker())

Each verdict is tiered and fanned out to the destinations its severity warrants — an escalation reaches the on-call officer as well as the log and scheduler, while an informational note stays in the log.

Failure modes and troubleshooting

Everything paged. Routing all findings to escalation drowns the on-call officer and desensitises them. Remediation: reserve escalation for regulatory violations and overdue warnings.
Blocking dispatch. Awaiting a slow channel inside the validation loop stalls the sweep. Remediation: publish to a bounded async queue drained by a worker pool.
Lost informational trail. Dropping informational alerts instead of logging them removes the trend data a fatigue review needs. Remediation: always log every tier, escalate selectively.
Hardcoded routes. Embedding destinations prevents re-pointing a tier during an incident. Remediation: drive routing from a configuration table.
No de-duplication. Re-emitting the same escalation on every sweep floods the officer. Remediation: de-duplicate by pairing and rule within a window before dispatch.

Frequently Asked Questions

How many severity tiers are enough?

Three — informational, warning, escalation — cover the operational need without overwhelming schedulers with granularity. More tiers rarely change who acts; they mostly add classification overhead. Start with three and split only if a real routing distinction demands it.

Why must dispatch be asynchronous?

Because alert delivery must never slow the validation sweep or block the scheduling UI. Publishing to a queue and draining it with workers decouples a slow escalation channel from the engine, so a paging outage cannot back-pressure compliance checks.

What separates a warning from an escalation?

A warning is a soft breach — a contractual buffer or policy — that the owning scheduler resolves before publication. An escalation is a hard regulatory violation, or a warning left unresolved past its deadline, that a compliance officer must see immediately.

Should informational alerts ever page someone?

No. Informational alerts are logged for trend review and never page anyone; that is what keeps the escalation channel meaningful. Elevating an informational note to a page is how alarm fatigue returns.

Threshold Tuning & Alerting — the parent section and alerting pipeline.
Tuning Alert Thresholds to Reduce Alarm Fatigue — the tuned levels this router consumes.
Building a Hash-Chained Audit Log in PostgreSQL — where every tier is durably recorded.
Rest Period Compliance Checks — a source of the verdicts being routed.

Back to Threshold Tuning & Alerting.