Tracking Dispute Resolution SLAs in Python

When the matching cascade hands an unresolved ledger entry to a human, the clock starts. Dispute-resolution Service Level Agreements (SLAs) define the contractual or regulatory window inside which a routed exception must reach a RESOLVED state before it counts as a breach. This page implements deterministic SLA tracking for those exceptions in Python: a frozen state record, monotonic countdown arithmetic that survives NTP and daylight-saving adjustments, a breach evaluator, and a cryptographically chained audit trail. It sits at the tail of Fallback Chain Configuration — once every automated resolution node is exhausted and the entry escalates into a manual queue, SLA tracking is what guarantees the escalation is acted on in time and is replayable for auditors.

Prerequisites

• The exception has already traversed the Fallback Chain Configuration nodes and emerged unresolved.
• A routing tier and risk weight are assigned by Threshold-Based Routing Logic, so the SLA window can be selected per tier.
• A destination Manual Review Queue exists for escalated items.
• Python 3.10+ with pydantic v2 installed; monetary fields use Decimal, never float.
• A structured logger is configured to emit trace_id, source_hash, and match_decision on every state change.

Step 1 — Model the dispute as a frozen state record

The record is immutable and strictly typed. Every transition produces a new instance rather than mutating in place, which keeps the audit chain honest. remaining_sla_seconds is never persisted — it is derived at query time to avoid clock drift across distributed workers.

from pydantic import BaseModel, ConfigDict, field_validator
from uuid import UUID
from decimal import Decimal
from datetime import datetime, timezone
from enum import Enum
import logging

log = logging.getLogger("sla.tracker")

class DisputeState(str, Enum):
    CREATED = "CREATED"
    ROUTED = "ROUTED"
    IN_REVIEW = "IN_REVIEW"
    ESCALATED = "ESCALATED"
    RESOLVED = "RESOLVED"
    LEDGER_POSTED = "LEDGER_POSTED"

class DisputeRecord(BaseModel):
    model_config = ConfigDict(frozen=True)

    dispute_id: UUID            # UUIDv7 recommended for temporal clustering
    ledger_batch_ref: str
    amount_usd: Decimal         # Decimal, not float, for monetary values
    sla_deadline_utc: datetime  # absolute, timezone-aware
    routing_tier: str
    state: DisputeState = DisputeState.CREATED
    source_hash: str            # hash of the originating unmatched payload
    prev_audit_hash: str | None = None

    @field_validator("amount_usd")
    @classmethod
    def two_dp(cls, v: Decimal) -> Decimal:
        if v <= 0:
            raise ValueError("amount_usd must be positive")
        return v.quantize(Decimal("0.01"))

Using Decimal is mandatory: float violates IEEE 754 precision guarantees for currency, and Pydantic v2’s decimal_places constraint is only valid on a Decimal field.

Step 2 — Compute the SLA deadline against a business calendar

The deadline is absolute and stored once. The window decrements against a jurisdiction-specific calendar that excludes weekends, market holidays, and banking cut-off times, so a 24-business-hour SLA opened on a Friday does not silently expire over a closed weekend.

from datetime import timedelta
from zoneinfo import ZoneInfo

def business_deadline(start_utc: datetime, sla_hours: int,
                      market: str = "America/New_York",
                      holidays: set[str] | None = None) -> datetime:
    holidays = holidays or set()
    tz = ZoneInfo(market)
    cursor = start_utc.astimezone(tz)
    remaining = timedelta(hours=sla_hours)
    while remaining > timedelta(0):
        cursor += timedelta(minutes=15)
        if cursor.weekday() >= 5:           # Sat/Sun
            continue
        if cursor.date().isoformat() in holidays:
            continue
        if not (9 <= cursor.hour < 17):     # trading window
            continue
        remaining -= timedelta(minutes=15)
    return cursor.astimezone(timezone.utc)

Step 3 — Track remaining time with a monotonic clock

For internal countdown comparisons use time.monotonic(), which never moves backwards. Reserve datetime.now(timezone.utc) for the reported timestamp on audit records. Anchoring the monotonic offset once per process and combining it with the stored UTC deadline gives drift-free remaining-time queries.

import time

class SlaClock:
    def __init__(self) -> None:
        self._wall0 = datetime.now(timezone.utc)
        self._mono0 = time.monotonic()

    def now_utc(self) -> datetime:
        elapsed = time.monotonic() - self._mono0
        return self._wall0 + timedelta(seconds=elapsed)

    def remaining_seconds(self, rec: DisputeRecord) -> float:
        return (rec.sla_deadline_utc - self.now_utc()).total_seconds()

Step 4 — Evaluate breach state and emit an audited transition

Each evaluation classifies the dispute as ON_TRACK, AT_RISK, or BREACHED, and every classification is logged with the audit triple. The priority score surfaces imminent breaches first so the Manual Review Queue drains the most time-sensitive items first.

import hashlib

def chain_hash(rec: DisputeRecord, decision: str, ts: datetime) -> str:
    blob = f"{rec.prev_audit_hash}|{rec.dispute_id}|{rec.state}|{decision}|{ts.isoformat()}"
    return hashlib.sha256(blob.encode("utf-8")).hexdigest()

def evaluate_sla(rec: DisputeRecord, clock: SlaClock,
                 risk_weight: float, at_risk_ratio: float = 0.20) -> str:
    total = (rec.sla_deadline_utc - clock._wall0).total_seconds()
    remaining = clock.remaining_seconds(rec)
    if remaining <= 0:
        decision = "BREACHED"
    elif remaining <= total * at_risk_ratio:
        decision = "AT_RISK"
    else:
        decision = "ON_TRACK"
    priority = (1.0 / max(remaining, 1.0)) * risk_weight
    ts = clock.now_utc()
    log.info(
        "sla.evaluated",
        extra={
            "trace_id": str(rec.dispute_id),
            "source_hash": rec.source_hash,
            "match_decision": decision,
            "remaining_seconds": round(remaining, 1),
            "priority_score": round(priority, 6),
            "audit_hash": chain_hash(rec, decision, ts),
        },
    )
    return decision

The append-only audit_hash chains each record to its predecessor, aligning with NIST SP 800-53 Rev. 5 AU-2 audit-event requirements and preventing retroactive SLA manipulation under SOX or GDPR review. Once an item reaches RESOLVED, the same chained-hash discipline carries through ledger posting, which is covered in Automating Batch Reconciliation Sign-Offs.

Configuration boundary table

Verification and testing

Use time-travel testing to simulate expiry without waiting in real time. A sample ledger fixture with a deadline pinned 30 minutes out lets you assert the classification boundaries deterministically.

def test_breach_boundary():
    clock = SlaClock()
    deadline = clock.now_utc() + timedelta(minutes=30)
    rec = DisputeRecord(
        dispute_id=UUID(int=7),
        ledger_batch_ref="BATCH-2026-06-26",
        amount_usd=Decimal("1499.005"),   # quantises to 1499.01
        sla_deadline_utc=deadline,
        routing_tier="standard",
        source_hash="a1b2c3",
    )
    assert rec.amount_usd == Decimal("1499.01")
    assert evaluate_sla(rec, clock, risk_weight=1.0) == "ON_TRACK"

    # Freeze the deadline in the past to force a breach.
    breached = rec.model_copy(update={"sla_deadline_utc": clock.now_utc() - timedelta(seconds=1)})
    assert evaluate_sla(breached, clock, risk_weight=1.0) == "BREACHED"

Run the evaluator as a jittered scheduled loop (Celery beat, APScheduler, or a Kubernetes CronJob) and export sla_remaining_seconds, breach_rate, and queue_depth as Prometheus gauges. Alert when breach_rate > 0.5% or p95_routing_latency > 200ms.

Troubleshooting

• SLA_CLOCK_SKEW — remaining time jumps between evaluations. Root cause: reading datetime.now() directly instead of the monotonic-anchored SlaClock, so NTP corrections leak into the countdown. Fix: derive every comparison from time.monotonic() and re-anchor on the monotonic_resync_s interval.
• NAIVE_DEADLINE — TypeError: can't subtract offset-naive and offset-aware datetimes. Root cause: a stored deadline lost its tzinfo. Fix: enforce timezone-aware values at the Pydantic boundary and persist as UTC.
• WEEKEND_FALSE_BREACH — items breach over a closed market. Root cause: a wall-clock countdown that ignores the trading calendar. Fix: compute the deadline via business_deadline() with the correct market and holidays set.
• DUPLICATE_ESCALATION — the same dispute escalates twice across workers. Root cause: no idempotency key on the transition. Fix: derive a key from dispute_id plus attempt_sequence and dedupe before queue insertion.
• FLOAT_AMOUNT_DRIFT — penny variances appear in reported totals. Root cause: amount_usd ingested as float. Fix: parse as Decimal and quantize(Decimal("0.01")) at validation, as in Step 1.

Related

• Setting Up Dynamic Routing Rules for High-Value Exceptions
• Designing a Slack-Integrated Approval Workflow for Unmatched Items
• Automating Batch Reconciliation Sign-Offs

Part of Fallback Chain Configuration within Exception Routing & Human-in-the-Loop Workflows.