IP Docketing Automation & Compliance

Building a Docket Escalation Ladder in Python

A docket escalation ladder is a small, deterministic state machine: it holds one unacknowledged deadline alert at a tier, starts an SLA timer, and on each evaluation either resolves the alert (a human acknowledged it), waits (the timer is still running), or promotes it to the next tier and restarts the timer. This page implements that machine in idiomatic Python 3.11+, with every transition sealed into a hash-chained audit ledger.

It is the runnable counterpart to the Escalation Routing Workflows design and consumes the unacknowledged reminders produced by the Deadline Reminder Cadence Engines upstream. The goal is a self-contained core you can drop into a scheduler tick and test without a database or a message bus.

What the Ladder Must Guarantee

Before any code, four properties fix the design. Promotion is forward-only — an alert climbs and never descends, so a deadline cannot ping-pong while its window closes. Only an attributable acknowledgment stops the ladder — not a delivery receipt, not elapsed time alone. Timers are absolute expiry instants, not in-memory countdowns, so a process restart resumes every in-flight escalation. And every transition is immutably logged to an append-only, hash-chained ledger, because the escalation record is the primary evidence in a missed-deadline inquiry. The implementation below encodes each property directly rather than leaving it to convention.

The escalation ladder evaluation loop for a single alert Each scheduler tick calls evaluate on an alert. First it checks whether the alert was acknowledged in its window; if so the alert is RESOLVED, an ACKNOWLEDGED entry is written, and the ladder stops. Otherwise it checks whether the SLA expiry has been reached; if not, it waits with a no-op until the next tick. If the expiry is reached, it checks whether the alert is already at the top tier; if not, it escalates by incrementing the tier, restarting the SLA timer, writing an ESCALATED entry, and looping back so the next tick re-evaluates the new tier. If it is already at the top tier, the alert fails closed to a manual backstop, writing a BREACHED_FINAL entry and triggering a client notice. Every emitted transition writes to an append-only, hash-chained ledger. evaluate(alert, now) one scheduler tick Acknowledged in window? SLA expiry reached? Already at top tier? ESCALATE tier += 1 · restart timer · emit ESCALATED RESOLVED emit ACKNOWLEDGED · stop WAIT no-op until next tick MANUAL BACKSTOP emit BREACHED_FINAL · notice no yes no yes no yes loop: next tick re-evaluates new tier Every ESCALATED / ACKNOWLEDGED / BREACHED_FINAL transition → append-only, hash-chained audit ledger (SHA-256)

Modelling Tiers and Alerts

Start with immutable, typed definitions. The ladder is a tuple of frozen tiers so it cannot be mutated at runtime, and each tier carries its role, its SLA as a timedelta, and the channels it may use. Modelling the role as a string that is later resolved against an assignment table — rather than a person — keeps the ladder stable across staffing changes.

from __future__ import annotations

import hashlib
from dataclasses import dataclass
from datetime import datetime, timedelta
from enum import Enum
from zoneinfo import ZoneInfo

UTC = ZoneInfo("UTC")


class AlertStatus(str, Enum):
    ACTIVE = "ACTIVE"                 # timer running at the current tier
    ACKNOWLEDGED = "ACKNOWLEDGED"     # terminal: a human accepted ownership
    BREACHED_FINAL = "BREACHED_FINAL"  # terminal: top tier lapsed → manual backstop


@dataclass(frozen=True)
class EscalationTier:
    role: str
    sla: timedelta
    channels: tuple[str, ...]


# Ordered ladder: paralegal → manager → attorney → group lead.
LADDER: tuple[EscalationTier, ...] = (
    EscalationTier("paralegal", timedelta(hours=8), ("email", "in_app")),
    EscalationTier("docketing_manager", timedelta(hours=4), ("email", "in_app", "sms")),
    EscalationTier("responsible_attorney", timedelta(hours=2), ("sms", "phone")),
    EscalationTier("practice_group_lead", timedelta(hours=1), ("sms", "page", "phone")),
)


@dataclass
class EscalatingAlert:
    """One deadline's live escalation state. `sla_expiry` is an absolute instant."""
    deadline_id: str
    tier_index: int
    sla_expiry: datetime
    status: AlertStatus = AlertStatus.ACTIVE

    @property
    def tier(self) -> EscalationTier:
        return LADDER[self.tier_index]

    @classmethod
    def open(cls, deadline_id: str, started: datetime) -> "EscalatingAlert":
        if started.tzinfo is None:
            raise ValueError("started must be timezone-aware")  # reject naive datetimes
        return cls(deadline_id, 0, started.astimezone(UTC) + LADDER[0].sla)

The Append-Only Ledger

The ledger is a hash-chained log. Each entry binds its fields and the previous entry’s hash into a SHA-256 digest, so reordering or back-dating any entry breaks the chain — the tamper-evidence a reviewer relies on. In production this backs onto write-once storage; here it is an in-memory list with the same contract, matching the immutable-ledger discipline of the Core Docketing Architecture & Deadline Types reference.

@dataclass
class LedgerEntry:
    deadline_id: str
    tier_role: str
    status: str
    actor: str
    occurred_utc: str
    prev_hash: str
    record_hash: str


class AppendOnlyLedger:
    def __init__(self) -> None:
        self._entries: list[LedgerEntry] = []
        self._tip = "0" * 64          # genesis hash

    def append(self, deadline_id: str, tier_role: str, status: str,
               actor: str, now: datetime) -> LedgerEntry:
        occurred = now.astimezone(UTC).isoformat()
        material = "|".join([deadline_id, tier_role, status, actor, occurred, self._tip])
        record_hash = hashlib.sha256(material.encode("utf-8")).hexdigest()
        entry = LedgerEntry(deadline_id, tier_role, status, actor,
                            occurred, self._tip, record_hash)
        self._entries.append(entry)
        self._tip = record_hash        # chain forward
        return entry

    @property
    def entries(self) -> list[LedgerEntry]:
        return list(self._entries)

The Evaluation Loop

evaluate is the whole state machine and maps one-to-one onto the flowchart above. It is pure with respect to time — it takes now as an argument rather than reading the clock — which is what makes it testable. A running timer is a no-op; a breach either promotes and restarts the timer or, at the top rung, fails closed.

def evaluate(alert: EscalatingAlert, now: datetime,
             ledger: AppendOnlyLedger) -> AlertStatus:
    """Advance one alert by a single tick. Forward-only; never descends."""
    if alert.status is not AlertStatus.ACTIVE:
        return alert.status                       # terminal — nothing to do
    if now < alert.sla_expiry:
        return alert.status                       # WAIT: timer still running

    # SLA breached without acknowledgment.
    if alert.tier_index + 1 < len(LADDER):
        breached_role = alert.tier.role
        alert.tier_index += 1                     # promote forward
        alert.sla_expiry = now + alert.tier.sla   # restart timer for the new tier
        ledger.append(alert.deadline_id, breached_role,
                      AlertStatus.ACTIVE.value + "->ESCALATED", "system", now)
        return alert.status                       # stays ACTIVE at the new tier

    # Top tier lapsed: fail closed into the manual backstop.
    alert.status = AlertStatus.BREACHED_FINAL
    ledger.append(alert.deadline_id, alert.tier.role,
                  AlertStatus.BREACHED_FINAL.value, "system", now)
    return alert.status


def acknowledge(alert: EscalatingAlert, who: str, now: datetime,
                ledger: AppendOnlyLedger) -> AlertStatus:
    """A human accepts ownership. Only an ACTIVE alert can be acknowledged."""
    if alert.status is not AlertStatus.ACTIVE:
        return alert.status
    alert.status = AlertStatus.ACKNOWLEDGED
    ledger.append(alert.deadline_id, alert.tier.role,
                  AlertStatus.ACKNOWLEDGED.value, who, now)
    return alert.status

Notice that acknowledge records who accepted and at which tier’s role, while an escalation is authored by "system". That distinction is what lets a reviewer see the difference between a human taking ownership and the machine promoting an alert nobody answered.

Running the Ladder

A short driver shows the intended flow: a paralegal never acknowledges, the alert escalates to the docketing manager after the 8-hour SLA, and the manager acknowledges within their window.

ledger = AppendOnlyLedger()
start = datetime(2026, 7, 16, 9, 0, tzinfo=UTC)
alert = EscalatingAlert.open("US-OA-4488", start)

# 9 hours later: the paralegal's 8h SLA lapsed → escalate to the manager.
evaluate(alert, start + timedelta(hours=9), ledger)
assert alert.tier.role == "docketing_manager"
assert alert.status is AlertStatus.ACTIVE

# The manager acknowledges an hour after that, inside their 4h window.
acknowledge(alert, "j.okafor", start + timedelta(hours=10), ledger)
assert alert.status is AlertStatus.ACKNOWLEDGED

for e in ledger.entries:
    print(e.tier_role, e.status, e.actor, e.occurred_utc)

The demo uses plain wall-clock hours for clarity. In production the SLA windows are measured in business hours against the shared office calendar, so an alert firing late on a Friday does not breach over a weekend — the same calendar the cadence engine and rule engine use, which keeps escalation timing defensible.

Operational Action: Take now as an argument everywhere and store sla_expiry as an absolute UTC instant. A durable scheduler then re-evaluates persisted alerts on restart, so a redeploy never silently forgets an in-flight escalation.

Testing

Because the machine is time-driven and forward-only, tests assert three things: a breach promotes, an acknowledgment stops the ladder, and the ledger stays chained.

import pytest
from datetime import datetime, timedelta


def test_breach_promotes_and_restarts_timer() -> None:
    led = AppendOnlyLedger()
    start = datetime(2026, 7, 16, 9, 0, tzinfo=UTC)
    a = EscalatingAlert.open("D-9", start)
    evaluate(a, start + timedelta(hours=9), led)     # 8h SLA lapsed
    assert a.tier_index == 1                         # promoted forward
    assert a.sla_expiry == start + timedelta(hours=9) + timedelta(hours=4)


def test_acknowledgment_is_terminal() -> None:
    led = AppendOnlyLedger()
    start = datetime(2026, 7, 16, 9, 0, tzinfo=UTC)
    a = EscalatingAlert.open("D-9", start)
    acknowledge(a, "m.ruiz", start + timedelta(hours=1), led)
    # A later tick must not resurrect or escalate a resolved alert.
    evaluate(a, start + timedelta(hours=99), led)
    assert a.status is AlertStatus.ACKNOWLEDGED
    assert a.tier_index == 0


def test_ledger_chain_is_linked() -> None:
    led = AppendOnlyLedger()
    start = datetime(2026, 7, 16, 9, 0, tzinfo=UTC)
    a = EscalatingAlert.open("D-9", start)
    evaluate(a, start + timedelta(hours=9), led)     # ESCALATED entry
    acknowledge(a, "m.ruiz", start + timedelta(hours=10), led)
    entries = led.entries
    assert len(entries) == 2
    assert entries[1].prev_hash == entries[0].record_hash


def test_naive_start_is_rejected() -> None:
    with pytest.raises(ValueError):
        EscalatingAlert.open("D-9", datetime(2026, 7, 16, 9, 0))  # no tzinfo

Frequently Asked Questions

Why does the evaluate function take the current time as an argument?
Because a function that reads the clock internally cannot be tested deterministically — you would have to sleep in real time to exercise an 8-hour SLA. Passing now in makes evaluate pure with respect to time, so a test can advance the clock instantly and assert the exact transition. It also lets a durable scheduler replay persisted alerts against a controlled instant on restart, which is what makes in-flight escalations survive a redeploy.
What stops a resolved alert from escalating again on a later tick?
Both evaluate and acknowledge early-return unless the alert is ACTIVE. Once an alert reaches a terminal status — ACKNOWLEDGED or BREACHED_FINAL — every subsequent tick is a no-op, so a resolved deadline can never be resurrected or promoted, even if the scheduler keeps ticking for days. The test that runs evaluate 99 hours after acknowledgment asserts exactly this.
Why record escalations as "system" but acknowledgments with a person's name?
The actor field is what lets a reviewer distinguish a human taking ownership from the machine promoting an alert nobody answered. An escalation is authored by the system because no person acted; an acknowledgment names the individual who accepted the deadline at that tier. That attribution is the core evidentiary value of the ledger — it shows not just that the process ran, but precisely who was accountable at each rung and when.

↑ Back to Escalation Routing Workflows

Related