SLA-Gated Escalation Rules for Aging Reconciliation Exceptions

An SLA-gated escalation engine is the control layer that decides when an unresolved exception stops belonging to its current reviewer and who it belongs to next. It runs after an item has already been routed and enqueued, watching the clock rather than the payload: as the remaining service-level window shrinks, the engine promotes the item up a tier ladder — junior reviewer to senior, senior to reconciliation desk, desk to managerial sign-off — so no break silently ages past a financial-close deadline. This engine sits deliberately above the mechanics of promotion. It decides the target tier and the trigger; the delivery of the promotion is handled by Fallback Chain Configuration, and the clock it reads is maintained by dispute-SLA tracking. This page operates as a direct extension of Exception Routing & Human-in-the-Loop Workflows, converting elapsed time and residual risk into a deterministic, auditable escalation decision a regulator can replay.

Prerequisites: Upstream Pipeline State

The escalation engine never re-evaluates matching and never re-derives materiality. It consumes decisions already made upstream and adds a single new dimension — time. Before an item is eligible for SLA gating, the following state must exist:

• A live SLA clock. Each exception must carry an sla_deadline and an accumulated business-hour elapsed value, both timezone-aware UTC. Computing that elapsed time correctly across weekends, holidays, and reviewer working hours is a problem in its own right, covered in Tracking Dispute Resolution SLAs in Python; this engine treats the resulting remaining-seconds figure as an immutable input.
• A resolved materiality band and risk tier. The amount classification and counterparty risk tier from Threshold-Based Routing Logic must already be attached. The escalation ladder is indexed by risk and materiality, so an unclassified item cannot be gated.
• A current review tier and assignment. The item’s present position on the ladder (TIER_1_REVIEWER, TIER_2_SENIOR, and so on) must be known, because escalation is always relative to where the item already sits.
• A globally unique reconciliation identifier and an escalation_epoch counter, both assigned at ingestion, so that repeated evaluations of the same aging item produce at most one promotion per epoch.

If any prerequisite is missing the item is held with a TIER_DEADEND code rather than promoted blindly, because escalating an exception whose clock or classification is unknown is itself a control gap.

The Escalation Ladder as a Directed Acyclic Graph

At its core the engine is a monotonic walk up a directed acyclic graph. Each node is an escalation tier with a distinct permission set, a distinct SLA budget, and a distinct notification channel; each edge is a legal promotion. The graph is acyclic and strictly upward — an item may skip a tier when severity demands it, but it may never demote, because a demotion would erase accountability that has already attached to a senior reviewer. The terminal node is managerial sign-off, from which the only exit is resolution.

Two ideas make the ladder deterministic. First, business-hour SLA windows: each tier owns an sla_seconds budget measured against the business calendar, not wall-clock, so an item raised at 16:00 Friday does not silently consume its four-hour window over a weekend. Second, at-risk versus breached bands: rather than acting only on a hard breach, the engine partitions each tier’s window into a healthy band, an at-risk band once elapsed time crosses at_risk_ratio × sla_seconds, and a breached band once the deadline passes. At-risk items are promoted proactively — surfacing to a senior reviewer before the deadline is what keeps the aggregate breach rate low — while breached items are promoted mandatorily and flagged for reporting.

Deterministic tier selection is a pure function of three axes: materiality × aging × risk. Materiality selects which ladder variant applies (a high-value exception uses a compressed ladder that reaches managerial review in fewer hops); aging measured in remaining SLA seconds selects the band; and risk tier sets how many rungs a single promotion may climb. Because the function is total and side-effect free, the same inputs always yield the same target tier — a property auditors rely on when they replay a decision months later.

The complexity of a single evaluation is O(1): computing the elapsed-to-budget ratio and indexing the ladder are constant-time operations, so the dominant cost when sweeping millions of open items is the durable read of each item’s clock, not the decision. The financial-domain caveat is that time is adversarial. Clock skew between the scheduler and the datastore, a daylight-saving transition, or a mis-configured holiday calendar can all make a healthy item look breached or a breached item look healthy — which is why the elapsed figure is always derived from the authoritative business-calendar source and never from the scheduler’s local clock.

Production-Grade Python Implementation

The evaluator below is the deterministic core. Given an item’s remaining SLA seconds and its risk tier, it selects the next escalation tier and emits a structured audit record carrying the trace_id, the source_hash, and the match_decision, so every promotion is independently replayable from the audit stream. Monetary and materiality figures are Decimal throughout; no float ever touches a comparison that could reclassify an item.

import hashlib
import logging
import uuid
from dataclasses import dataclass
from decimal import Decimal, ROUND_HALF_UP
from datetime import datetime, timezone
from enum import IntEnum
from typing import Optional
from pydantic import BaseModel, Field, field_validator

logger = logging.getLogger("reconciliation.escalation")


class Tier(IntEnum):
    TIER_1_REVIEWER = 1
    TIER_2_SENIOR = 2
    TIER_3_DESK = 3
    MANAGERIAL = 4


class Band(str):
    HEALTHY = "HEALTHY"
    AT_RISK = "AT_RISK"
    BREACHED = "BREACHED"


@dataclass(frozen=True)
class TierRule:
    sla_seconds: int
    at_risk_ratio: Decimal          # fraction of budget at which we pre-promote
    escalation_cooldown_s: int      # min seconds between promotions of one item
    notify_channel: str


# Ordered ladder; index by Tier. MANAGERIAL is terminal (no rule beyond it).
LADDER: dict[Tier, TierRule] = {
    Tier.TIER_1_REVIEWER: TierRule(14400, Decimal("0.70"), 900, "#recon-review"),
    Tier.TIER_2_SENIOR:   TierRule(7200,  Decimal("0.75"), 600, "#recon-senior"),
    Tier.TIER_3_DESK:     TierRule(3600,  Decimal("0.80"), 300, "pager-desk"),
}


class Exception(BaseModel):
    item_id: str = Field(default_factory=lambda: str(uuid.uuid4()))
    reconciliation_id: str
    current_tier: Tier
    risk_tier: int                  # 1 (low) .. 5 (severe), from routing
    materiality: Decimal
    match_decision: str             # MATCHED | PARTIAL | UNMATCHED, from upstream
    escalation_epoch: int = 0
    last_escalated_at: Optional[datetime] = None
    trace_id: str = Field(default_factory=lambda: str(uuid.uuid4()))

    @field_validator("materiality", mode="before")
    @classmethod
    def enforce_precision(cls, v) -> Decimal:
        return Decimal(str(v)).quantize(Decimal("0.01"), rounding=ROUND_HALF_UP)


def classify_band(elapsed_s: int, rule: TierRule) -> str:
    if elapsed_s >= rule.sla_seconds:
        return Band.BREACHED
    if Decimal(elapsed_s) >= rule.at_risk_ratio * Decimal(rule.sla_seconds):
        return Band.AT_RISK
    return Band.HEALTHY


def select_target_tier(current: Tier, risk_tier: int) -> Optional[Tier]:
    # Higher risk climbs more rungs per promotion, but never past terminal.
    hops = 2 if risk_tier >= 4 else 1
    target = Tier(min(int(current) + hops, int(Tier.MANAGERIAL)))
    return None if target == current else target


def _source_hash(exc: Exception, band: str, target: Tier) -> str:
    material = (
        f"{exc.item_id}:{exc.reconciliation_id}:{exc.current_tier.name}:"
        f"{target.name}:{band}:{exc.materiality}:{exc.escalation_epoch}"
    )
    return hashlib.sha256(material.encode("utf-8")).hexdigest()


def evaluate(exc: Exception, elapsed_s: int, now: datetime) -> Optional[Tier]:
    """Return the tier to promote to, or None if no escalation is due.

    Idempotent per (item, epoch): a second evaluation inside the cooldown,
    or one that recomputes the same target, is a no-op.
    """
    rule = LADDER.get(exc.current_tier)
    if rule is None:                       # already terminal
        return None

    band = classify_band(elapsed_s, rule)
    if band == Band.HEALTHY:
        return None

    # Single-escalation guard: honour the per-item cooldown window.
    if exc.last_escalated_at is not None:
        since = (now - exc.last_escalated_at).total_seconds()
        if since < rule.escalation_cooldown_s:
            match_decision = "cooldown_suppressed"
            logger.info(
                "escalation_suppressed",
                extra={
                    "trace_id": exc.trace_id,
                    "source_hash": _source_hash(exc, band, exc.current_tier),
                    "match_decision": exc.match_decision,
                    "item_id": exc.item_id,
                    "band": band,
                    "reason": "cooldown",
                    "seconds_since_last": since,
                },
            )
            return None

    target = select_target_tier(exc.current_tier, exc.risk_tier)
    if target is None:
        return None

    source_hash = _source_hash(exc, band, target)
    # Commit the escalation state atomically with the audit emission.
    exc.escalation_epoch += 1
    exc.last_escalated_at = now
    prior, exc.current_tier = exc.current_tier, target

    logger.info(
        "escalation_promoted",
        extra={
            "trace_id": exc.trace_id,
            "source_hash": source_hash,
            "match_decision": exc.match_decision,
            "item_id": exc.item_id,
            "reconciliation_id": exc.reconciliation_id,
            "prior_tier": prior.name,
            "target_tier": target.name,
            "band": band,
            "risk_tier": exc.risk_tier,
            "materiality": str(exc.materiality),
            "notify_channel": LADDER.get(target, TierRule(0, Decimal(0), 0,
                                                          "managerial")).notify_channel,
            "epoch": exc.escalation_epoch,
            "ts": now.astimezone(timezone.utc).isoformat(),
        },
    )
    return target

The escalation_epoch and the cooldown together are the idempotent single-escalation guard: they guarantee that a scheduler firing twice within the same window — a retry storm, a duplicated timer, two overlapping sweep workers — promotes an item at most once. The source_hash binds the decision to the exact inputs, so two workers evaluating the same aging item derive identical audit records and can never emit two conflicting promotions.

Configuration Rules & Threshold Calibration

Every tier parameter is version-controlled and timestamped; changing one is itself an auditable event. Configuration is hot-reloadable from a versioned YAML manifest validated against JSON Schema before deployment, so the ladder can absorb a regulatory change or a quarter-close volume spike without a service restart. Treat the defaults below as conservative starting points and widen or compress them only with evidence from replayed historical exceptions.

Calibration is empirical, not intuitive. Replay a representative window of closed exceptions through the engine, plot the realised breach rate against at_risk_ratio, and choose the ratio where earlier promotion stops meaningfully reducing breaches — beyond that point you are only manufacturing noise for senior reviewers. Document the chosen envelope alongside the reconciliation run so an auditor can reconstruct why a given item escalated when it did.

Multi-Dimensional Validation

A single elapsed-time check is not sufficient to justify a promotion. Robust gating composes complementary constraints and escalates only when the combination crosses the boundary — the logical intersection of independent gates, never a weighted average that lets one dimension mask another:

1. Temporal gate — the business-hour elapsed value against the tier’s sla_seconds, evaluated through classify_band, as the primary trigger.
2. Materiality and risk gate — the Decimal materiality band and counterparty risk tier from Threshold-Based Routing Logic, which decides how many rungs a single promotion climbs and which ladder variant applies.
3. Ladder-integrity gate — a structural check that the proposed target is reachable and strictly above the current tier, so a corrupt tier value can never produce a demotion or a leap past managerial sign-off.

Because the gates are independent, partial satisfaction is meaningful: a low-materiality item that breaches its window climbs one rung, while a severe-risk item that is merely at-risk may climb two — the same clock, different destinations. The engine records which gates fired so the promotion is explainable after the fact.

Async / Scheduler Execution Patterns

At scale the engine is a periodic sweep, not a per-item thread. A single scheduler wakes on a fixed cadence, streams the open-exception set from the durable store using FOR UPDATE SKIP LOCKED so overlapping sweeps never contend on the same row, and evaluates each item on a non-blocking asyncio loop. Deterministic tier selection is CPU-bound and trivially fast, so the loop’s real constraint is I/O to the datastore and to notification channels — both of which are awaited, not blocked on.

import asyncio
from datetime import datetime, timezone
from typing import AsyncIterator


async def sweep_worker(
    name: str,
    due: "asyncio.Queue[tuple[Exception, int]]",
    notify: "asyncio.Queue[tuple[Exception, Tier]]",
) -> None:
    while True:
        exc, elapsed_s = await due.get()
        try:
            now = datetime.now(timezone.utc)
            target = await asyncio.to_thread(evaluate, exc, elapsed_s, now)
            if target is not None:
                await notify.put((exc, target))   # backpressure on notify tier
        except Exception as err:  # noqa: BLE001 - isolate one bad item
            logger.exception("escalation_eval_failed",
                             extra={"item_id": exc.item_id, "err": str(err)})
        finally:
            due.task_done()


async def run_sweep(
    open_items: AsyncIterator["tuple[Exception, int]"],
    concurrency: int = 8,
    max_inflight: int = 2000,
) -> None:
    due: "asyncio.Queue[tuple[Exception, int]]" = asyncio.Queue(maxsize=max_inflight)
    notify: "asyncio.Queue[tuple[Exception, Tier]]" = asyncio.Queue(maxsize=max_inflight)
    workers = [
        asyncio.create_task(sweep_worker(f"sweep-{i}", due, notify))
        for i in range(concurrency)
    ]
    async for exc, elapsed_s in open_items:  # locked stream via SKIP LOCKED
        await due.put((exc, elapsed_s))       # producer blocks when saturated
    await due.join()
    for w in workers:
        w.cancel()

The actual promotion — reassigning the item and delivering the notification — is handed to Fallback Chain Configuration, which owns the tier permissions and the delivery guarantees. Keeping the decision and the delivery in separate components is what lets the engine be a pure function: it can be replayed, unit-tested, and audited without ever sending a real page. The durable timer mechanics that wake the sweep at exactly the right moment for each item — rather than polling the entire open set every cycle — are covered in Scheduling Escalations with Redis Sorted Sets, which uses a score-ordered set keyed on each item’s next SLA boundary.

Failure Modes & Remediation

The engine carries its own catalogue of named failure codes. Each is emitted with the trace_id and source_hash so it can be triaged from the audit stream rather than reproduced by hand.

The unifying remediation principle is that no failure silently absorbs an item: every code either promotes, holds, or returns an idempotent no-op, and none ever demotes. A NOTIFY_FAILED in particular must never be treated as “escalation complete” — an undelivered page is an unresolved breach, and the item stays gated until a channel confirms delivery.

Compliance & Audit Trail Requirements

Financial reconciliation systems must satisfy SOX Section 404, PCI-DSS Requirement 10, and GAAP close controls. Because escalation decides who is accountable for an aging break, its audit trail is not optional evidence — it is the segregation-of-duties record. The engine enforces compliance through three mechanisms, and every promotion emits a record carrying its trace_id, source_hash, and match_decision for SOX traceability:

1. Immutable escalation ledger. Every promotion, suppression, and breach is serialised to a WORM tier capturing actor, timestamp, prior and target tier, band, and a cryptographic hash of the inputs. Cryptographic chaining prevents retroactive edits, and each event is replayable for forensic reconstruction of why an item reached managerial review when it did.
2. Deterministic replayability. Because tier selection is a pure function of materiality × aging × risk, an auditor can re-run the engine against the frozen inputs and obtain the identical target tier — closing the gap between “the system escalated it” and “the system was correct to escalate it”.
3. Breach reporting. Every SLA_BREACH is timestamped against the business calendar and surfaced in close-period reporting, so an exception that ages past a regulatory window is disclosed rather than buried. Threshold manifests are version-controlled, letting the system reconstruct the exact SLA envelope active at execution time.

By treating SLA-gated escalation as a deterministic, replayable decision layer sitting above the fallback-chain machinery and the dispute-SLA clock, FinOps and accounting-engineering teams ensure that no reconciliation break silently outlives its window — and that every promotion up the ladder is an auditable, defensible event.

Related

• Fallback Chain Configuration
• Tracking Dispute Resolution SLAs in Python
• Threshold-Based Routing Logic
• Scheduling Escalations with Redis Sorted Sets

Part of Exception Routing & Human-in-the-Loop Workflows.