Approval SLA Monitoring: Deterministic Time-in-State Timers for Reimbursement Queues

Approval SLA monitoring is the measurement layer that turns “where is my reimbursement stuck?” into a numeric, alertable signal — it clocks how long every report has held its current approval state, compares that elapsed time against a budget defined per approval tier, and emits a breach signal the moment a report ages past its allowance. Within the broader Reimbursement Routing & Approval Sync framework, this component observes the same append-only ledger that routing writes to, computes time-in-state deterministically, and produces the aging buckets and breach metrics that downstream escalation acts on. It owns measurement and thresholding; it delegates the actual re-assignment of a stalled report to Approval Chain Routing, which consumes the breach signals this component emits. This guide covers the timer engine, its budget configuration surface, and the immutable metrics trail that makes every SLA decision reconstructable months later during an audit.

The engineering objective is not a dashboard that looks busy — it is a deterministic clock, where the same ledger replayed twice yields the identical breach set, so an alert can never be blamed on measurement jitter. Everything below is built to make the measurement reproducible and the threshold explicit.

Problem Framing & Root Causes

The dominant production failure is the silent stall: a report enters PENDING_APPROVAL, the assigned approver is on leave, and nothing pages anyone until the employee complains three weeks later. Naive monitors miss this because they measure the wrong clock. Four named failure modes account for almost every missed SLA in an accounts-payable queue.

Wall-clock drift. Timers computed from submitted_at rather than the timestamp of the current state transition count time the report spent in states it already cleared. A report that sat one day in SUBMITTED, cleared manager review, and has now been in FINANCE_REVIEW for an hour is not “26 hours late” — but a submitted-at timer says it is, and the resulting false page trains reviewers to ignore alerts.

Business-hours blindness. A 24-hour budget that ticks through weekends and holidays fires on Monday morning for a report submitted Friday at 5 p.m., even though no approver was ever expected to act. SLA budgets that are not expressed against a business calendar generate a predictable Monday alert storm that buries the genuine stalls.

Tier flattening. A single global budget treats a junior approver’s routine sign-off and a CFO’s high-value exception the same. The CFO tier legitimately needs longer; the junior tier should page fast. Without a per-tier budget, one threshold is simultaneously too aggressive for one queue and too lax for another.

Non-monotonic re-entry. When a report is rejected back to an earlier state and re-approved, a monitor that keys on “first entered PENDING_APPROVAL” double-counts the earlier wait. Time-in-state must reset on every transition into the state and measure only the current, contiguous dwell.

The remedy is a timer that computes elapsed time from the last transition into the current state, evaluates it against a calendar-aware budget selected by approval tier, and treats every evaluation as an immutable, replayable record.

Design Constraints & Prerequisites

This component is read-mostly against the routing ledger and write-only against its own metrics ledger; it never mutates a report’s approval state, which keeps it safe to run as a sidecar cron without racing the routing engine. It assumes upstream routing has already emitted a well-formed transition event for every state change, carrying the report id, the state entered, the approval tier, and a timezone-aware UTC timestamp. If a transition event is malformed or missing its tier, the report must divert to a review lane rather than be silently assigned the default budget.

The tier-to-budget table and the business calendar are the only shared configuration; treat both as configuration-as-code so finance can retune an SLA without a redeploy. Because escalation is a separate concern, this component emits a structured breach signal and stops there — the re-assignment logic lives in Approval Chain Routing.

Production Python Implementation

The engine is three composable stages: a deterministic time-in-state calculator that folds the ledger, an aging classifier plus budget gate that turns elapsed time into a decision, and a metrics-and-audit emitter. Each is independently testable and takes an injected clock so its output is reproducible.

Deterministic time-in-state from the ledger

The calculator folds a report’s ordered transition history down to a single fact: the state it currently holds and the instant it entered that state. Elapsed time is now() - entered_at, measured against an injected clock so a test or a replay produces the identical value.

from __future__ import annotations

import logging
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
from typing import Callable, Iterable

logger = logging.getLogger("expense.sla.timer")


@dataclass(frozen=True)
class TransitionEvent:
    """One append-only ledger row: a report entered `to_state` at `event_ts`."""
    report_id: str
    to_state: str
    approval_tier: str
    event_ts: datetime  # timezone-aware UTC


@dataclass(frozen=True)
class StateDwell:
    """The report's current state and how long it has held it."""
    report_id: str
    current_state: str
    approval_tier: str
    entered_at: datetime
    elapsed: timedelta


def compute_dwell(
    events: Iterable[TransitionEvent],
    now: Callable[[], datetime],
) -> StateDwell:
    """Fold a report's transition history into its current time-in-state.

    Elapsed is measured only from the LATEST transition into the current
    state, so a rejection-then-reapproval loop never double-counts earlier
    dwell. `now` is injected to keep the result deterministic under replay.
    """
    ordered = sorted(events, key=lambda e: e.event_ts)
    if not ordered:
        raise ValueError("cannot compute dwell for an empty transition history")

    latest = ordered[-1]
    if latest.event_ts.tzinfo is None:
        raise ValueError(f"event_ts must be tz-aware for report {latest.report_id}")

    current = now().astimezone(timezone.utc)
    elapsed = current - latest.event_ts
    if elapsed < timedelta(0):
        # Clock skew or a future-dated event — clamp and flag, never negative.
        logger.warning("negative_elapsed_clamped", extra={"report_id": latest.report_id})
        elapsed = timedelta(0)

    dwell = StateDwell(
        report_id=latest.report_id,
        current_state=latest.to_state,
        approval_tier=latest.approval_tier,
        entered_at=latest.event_ts,
        elapsed=elapsed,
    )
    logger.debug(
        "dwell_computed",
        extra={"report_id": dwell.report_id, "state": dwell.current_state,
               "elapsed_s": elapsed.total_seconds()},
    )
    return dwell

Folding the ledger rather than trusting a mutable entered_at column means the timer is derived purely from immutable events; if the ledger is intact, the elapsed value is provable. Reports whose latest event lacks a tier or a timezone raise here and must divert to a review lane rather than reach the gate.

Aging classifier and per-tier budget gate

Elapsed time alone is not a decision. The classifier converts it into an aging bucket for reporting, then the gate compares it against the business-hours budget for the report’s tier and returns an explicit WITHIN_SLA or SLA_BREACH verdict. Budgets are expressed in business hours and translated through a simple calendar so weekends never tick.

from __future__ import annotations

from dataclasses import dataclass
from datetime import datetime, timedelta
from enum import Enum
from typing import Mapping


class AgingBucket(str, Enum):
    FRESH = "FRESH"
    WARNING = "WARNING"
    OVERDUE = "OVERDUE"


@dataclass(frozen=True)
class SlaBudget:
    """Business-hours budget for one approval tier, with a warn fraction."""
    tier: str
    budget_business_hours: float
    warn_fraction: float = 0.75  # bucket flips to WARNING at this share of budget


@dataclass(frozen=True)
class SlaVerdict:
    report_id: str
    approval_tier: str
    current_state: str
    business_hours_elapsed: float
    budget_business_hours: float
    bucket: AgingBucket
    breached: bool


def business_hours_between(start: datetime, end: datetime,
                           work_start: int = 9, work_end: int = 17) -> float:
    """Count business hours (Mon-Fri, work_start..work_end) between two UTC instants.

    Deterministic and calendar-aware so weekend dwell does not accrue budget.
    Holidays can be layered by subtracting matched work days upstream.
    """
    if end <= start:
        return 0.0
    total = 0.0
    cursor = start
    step = timedelta(minutes=15)
    span_hours = (work_end - work_start)
    while cursor < end:
        if cursor.weekday() < 5 and work_start <= cursor.hour < work_end:
            total += step.total_seconds() / 3600.0
        cursor += step
    # Guard against runaway accumulation on absurd inputs.
    return min(total, span_hours * 5 * 520)


def evaluate_sla(dwell, now, budgets: Mapping[str, SlaBudget]) -> SlaVerdict:
    """Classify aging and decide WITHIN_SLA vs SLA_BREACH for a report."""
    budget = budgets.get(dwell.approval_tier)
    if budget is None:
        raise KeyError(f"no SLA budget configured for tier {dwell.approval_tier!r}")

    elapsed_bh = business_hours_between(dwell.entered_at, now())
    ratio = elapsed_bh / budget.budget_business_hours if budget.budget_business_hours else 1.0

    if ratio >= 1.0:
        bucket = AgingBucket.OVERDUE
    elif ratio >= budget.warn_fraction:
        bucket = AgingBucket.WARNING
    else:
        bucket = AgingBucket.FRESH

    return SlaVerdict(
        report_id=dwell.report_id,
        approval_tier=dwell.approval_tier,
        current_state=dwell.current_state,
        business_hours_elapsed=round(elapsed_bh, 2),
        budget_business_hours=budget.budget_business_hours,
        bucket=bucket,
        breached=bucket is AgingBucket.OVERDUE,
    )

Expressing the budget in business hours is what eliminates the Monday alert storm: a report submitted Friday afternoon accrues no budget over the weekend, so it is not spuriously overdue at Monday open. The warn_fraction gives operations a heads-up bucket before the hard breach, which is what lets a team nudge an approver before an SLA is formally missed.

Metrics and audit emission

Every evaluation — pass or breach — emits a structured time-series metric and appends one immutable row to the SLA ledger, so the aging distribution is a first-class monitored signal and any single breach is reconstructable from its inputs. Only a SLA_BREACH verdict produces the escalation signal that routing consumes; a WITHIN_SLA verdict is logged and metered but takes no action.

from __future__ import annotations

import hashlib
import json
import logging
from datetime import datetime, timezone
from typing import Any, Protocol

logger = logging.getLogger("expense.sla.emit")


class MetricSink(Protocol):
    def observe(self, *, name: str, value: float, tags: dict[str, str]) -> None: ...


class EscalationSink(Protocol):
    def signal_breach(self, *, report_id: str, tier: str, overdue_by_bh: float) -> None: ...


def emit_verdict(verdict, metrics: MetricSink, escalation: EscalationSink,
                 now) -> dict[str, Any]:
    """Meter the verdict, append an audit row, and signal escalation on breach.

    The audit row carries a SHA-256 over the inputs that produced the verdict,
    so a stored decision can be re-derived and proven during a control review.
    """
    metrics.observe(
        name="approval_sla_elapsed_business_hours",
        value=verdict.business_hours_elapsed,
        tags={"tier": verdict.approval_tier, "state": verdict.current_state,
              "bucket": verdict.bucket.value},
    )

    payload = {
        "report_id": verdict.report_id,
        "approval_tier": verdict.approval_tier,
        "current_state": verdict.current_state,
        "business_hours_elapsed": verdict.business_hours_elapsed,
        "budget_business_hours": verdict.budget_business_hours,
        "bucket": verdict.bucket.value,
        "verdict": "SLA_BREACH" if verdict.breached else "WITHIN_SLA",
        "evaluated_at": now().astimezone(timezone.utc).isoformat(),
    }
    audit_row = dict(payload)
    audit_row["compliance_hash"] = hashlib.sha256(
        json.dumps(payload, sort_keys=True).encode("utf-8")
    ).hexdigest()

    if verdict.breached:
        overdue_by = verdict.business_hours_elapsed - verdict.budget_business_hours
        metrics.observe(name="approval_sla_breach_total", value=1.0,
                        tags={"tier": verdict.approval_tier})
        escalation.signal_breach(report_id=verdict.report_id,
                                 tier=verdict.approval_tier,
                                 overdue_by_bh=round(overdue_by, 2))
        logger.warning("sla_breach", extra=audit_row)
    else:
        logger.info("sla_within", extra=audit_row)
    return audit_row

The compliance_hash binds the verdict to the exact elapsed value and budget that produced it, so an auditor replaying the ledger can prove the breach fired for the right reason. The escalation sink is deliberately thin — it hands off to Approval Chain Routing and returns, keeping measurement and re-assignment cleanly separated.

Configuration Reference

Expose every threshold through a centralized policy store or environment so finance can retune SLAs without a code change. Keep the tier budget table and business calendar versioned; a silent budget change is an unlogged control change.

Validation & Testing

Test the clock, not the plumbing: assert that identical ledgers under a fixed now() yield identical elapsed values, that weekend dwell accrues no budget, and that a rejection loop does not double-count. Inject the clock so every case is deterministic.

from datetime import datetime, timezone


def _fixed(ts: datetime):
    return lambda: ts


def test_dwell_measures_only_current_state_entry():
    events = [
        TransitionEvent("R1", "SUBMITTED", "manager",
                        datetime(2026, 7, 13, 9, 0, tzinfo=timezone.utc)),
        TransitionEvent("R1", "PENDING_APPROVAL", "manager",
                        datetime(2026, 7, 14, 9, 0, tzinfo=timezone.utc)),
    ]
    dwell = compute_dwell(events, _fixed(datetime(2026, 7, 14, 12, 0, tzinfo=timezone.utc)))
    assert dwell.current_state == "PENDING_APPROVAL"
    assert dwell.elapsed.total_seconds() == 3 * 3600  # only the 3h in the latest state


def test_weekend_accrues_no_budget():
    budgets = {"manager": SlaBudget("manager", budget_business_hours=16)}
    entered = datetime(2026, 7, 17, 16, 0, tzinfo=timezone.utc)  # Friday 16:00
    dwell = StateDwell("R2", "PENDING_APPROVAL", "manager", entered,
                       elapsed=None)  # elapsed unused by evaluate_sla
    monday_open = datetime(2026, 7, 20, 9, 0, tzinfo=timezone.utc)
    verdict = evaluate_sla(dwell, _fixed(monday_open), budgets)
    assert verdict.bucket is AgingBucket.FRESH  # only ~1 business hour Fri accrued


def test_breach_when_over_tier_budget():
    budgets = {"manager": SlaBudget("manager", budget_business_hours=8)}
    entered = datetime(2026, 7, 13, 9, 0, tzinfo=timezone.utc)  # Monday 09:00
    dwell = StateDwell("R3", "PENDING_APPROVAL", "manager", entered, elapsed=None)
    verdict = evaluate_sla(dwell, _fixed(datetime(2026, 7, 14, 15, 0, tzinfo=timezone.utc)), budgets)
    assert verdict.breached and verdict.bucket is AgingBucket.OVERDUE

Gate deployments on these assertions in CI. Add fixtures for the missing-tier divert and for a report re-entering PENDING_APPROVAL after rejection, confirming the second entry resets the clock rather than summing both waits.

Operational Runbook

1. Seed the tier budgets. Populate SLA_TIER_BUDGETS from the approval matrix finance already maintains, and back it with the same business calendar payroll uses, so budgets and holidays agree from day one.
2. Run the sweep as a stateless sidecar. Schedule the monitor every SLA_SWEEP_INTERVAL_S; it reads open reports, folds each report’s ledger via compute_dwell, and evaluates — it never writes approval state, so it cannot race routing.
3. Deduplicate on state entry. Key the breach ledger on (report_id, entered_at) so a rerun after a crash re-evaluates but appends at most one breach row per state entry.
4. Set alert thresholds on the metrics, not the raw queue. Page when the OVERDUE bucket share exceeds its trailing baseline, when approval_sla_breach_total for any tier spikes, or when the WARNING bucket grows faster than it drains — an early sign approvers are falling behind.
5. Hand breaches to escalation, not to a human inbox. Route every SLA_BREACH signal into Approval Chain Routing; the monitor’s job ends at emitting the signal and the audit row.
6. Retune with evidence. When a budget is changed, version the config and annotate the metrics timeline; the compliance_hash on each prior row keeps historical breaches interpretable under the old budget even after the new one lands.

Approval SLA monitoring is a deterministic measurement control, not a heuristic dashboard. Ledger-derived timers, per-tier business-hours budgets, and immutable metrics give finance a component that fires the right alert at the right time, survives replay unchanged, and produces the breach signals escalation needs to keep reimbursements moving.

Related

• Reimbursement Routing & Approval Sync — the parent framework this measurement layer plugs into
• Detecting stuck approvals with SLA timers — the focused scanner that acts on these timers to catch stalled reports
• Approval Chain Routing — consumes the breach signals emitted here to re-assign and escalate
• ERP Export Synchronization — the downstream stage that receives reports once approvals clear on time
• Automated Policy Validation & Anomaly Flagging — the validation framework whose outcomes gate a report into the approval queue