Spending Cap Hierarchies: Deterministic Ceiling Resolution for Expense Pipelines

Spending cap hierarchies resolve the many overlapping monetary limits that apply to a single expense line — corporate, departmental, role, category, and location — into one binding ceiling and one defensible verdict per transaction.

Within the broader Core Policy Architecture & Taxonomy Design control plane, this component owns one job: given a normalized, already-categorized expense line, compute the tightest applicable limit and flag any breach with an auditable resolution trail. It does not assign the expense category — that is delegated to Expense Category Taxonomies — nor does it compute the location-aware daily allowance itself, which is produced upstream by Per Diem Rate Structuring and consumed here as one input tier. It assumes the record already cleared confidence gating in the Receipt Ingestion & OCR Data Extraction framework. This page covers the precedence model, the streaming resolver, its configuration surface, and the audit metadata that make every ceiling decision replayable. The concrete engine walkthrough lives in implementing tiered spending caps in Python.

Problem Framing & Root Causes

The core failure is ceiling ambiguity: a lodging line for a director in Zurich is simultaneously subject to a corporate hard cap, a departmental travel envelope, a role limit, a lodging-category ceiling, and a per-diem override — and naive code either picks one arbitrarily or applies the wrong one. Three named failure modes follow:

• Precedence drift — resolution order lives implicitly in evaluation sequence, so refactoring the loop silently changes which cap binds and re-audits a historical report to a different verdict.
• Mutable-snapshot contamination — rules are read from a live table mid-batch, so a policy edit at 02:14 makes lines 400,000–500,000 audit against different limits than lines 1–399,999 in the same run.
• Materialization thrash — loading a full million-row ledger into one in-memory frame triggers garbage-collection stalls and non-deterministic latency, defeating the guarantee that identical input yields identical timing and output.

Design Constraints & Prerequisites

The resolver sits after categorization and before approval routing, and inherits hard contracts from both sides. It must only run on records that already carry a stable category node, a coerced amount as Decimal, and a resolved location key; introducing it earlier amplifies OCR noise, and delaying it past routing lets non-compliant spend reach settlement.

Concrete prerequisites:

• Upstream data contract: every line carries line_id, category, role_level, department, location_key, amount (Decimal), and submission_date. Missing fields fail schema validation rather than defaulting silently.
• Immutable policy snapshot: the full cap set is loaded once per run, hashed, and frozen. No cap is read from a mutable store during evaluation, so the whole batch audits against one policy_version.
• Determinism: resolution is a pure function of the line and the snapshot — no wall-clock branching, no network calls, no floating-point money.
• Throughput / memory: monthly batches routinely exceed a million lines, so the resolver streams bounded chunks and never materializes the ledger. Per-chunk work is stateless; only aggregate counters and the append-only audit log persist between chunks.
• Compliance precondition: every decision emits an audit event carrying the policy version and a deterministic decision hash, so a verdict is replayable years later under SOX and internal-audit review.

The precedence order is fixed and explicit, from hardest to softest constraint. Every applicable tier is evaluated; the binding ceiling is the smallest limit across all of them, and ties resolve to the tier with the lowest ordinal so the resolution is total and deterministic:

1. Absolute corporate cap — hard ceiling, never overridden.
2. Department / project cap — budget-bound envelope.
3. Role / level cap — seniority-adjusted personal limit.
4. Category-specific cap — lodging vs. meals vs. ground transport.
5. Per-diem / geographic override — location-aware daily allowance from the per-diem layer.

Production Python Implementation

The resolver below is self-contained and runnable. It models caps as frozen, tier-tagged rules with scope selectors, computes the binding ceiling as the minimum across all applicable tiers, and emits a SHA-256 decision hash plus the full resolution path for every line. Money is Decimal throughout to keep rounding deterministic. Ingestion is generator-based so memory stays flat regardless of batch size.

from __future__ import annotations

import hashlib
import json
import logging
from dataclasses import dataclass, field
from datetime import datetime, timezone
from decimal import Decimal, ROUND_HALF_UP
from enum import IntEnum
from typing import Iterable, Iterator, Optional

# Structured, append-only audit logging aligned with SOX retention.
logging.basicConfig(level=logging.INFO, format="%(message)s")
audit_logger = logging.getLogger("policy.spending_caps")


class CapTier(IntEnum):
    """Precedence order. Lower ordinal = harder constraint; wins ties."""
    ABSOLUTE = 1      # corporate hard ceiling, never overridden
    DEPARTMENT = 2    # budget-bound department/project envelope
    ROLE = 3          # seniority-adjusted personal limit
    CATEGORY = 4      # per-category ceiling (lodging vs meals)
    PER_DIEM = 5      # location-aware daily allowance


@dataclass(frozen=True)
class Cap:
    """A single ceiling and the scope it applies to.

    A selector field set to None means "matches any value" for that
    dimension, so a corporate ABSOLUTE cap leaves every selector None.
    """
    cap_id: str
    tier: CapTier
    limit: Decimal
    category: Optional[str] = None
    role_level: Optional[str] = None
    department: Optional[str] = None
    location_key: Optional[str] = None

    def applies_to(self, line: "ExpenseLine") -> bool:
        return (
            (self.category is None or self.category == line.category)
            and (self.role_level is None or self.role_level == line.role_level)
            and (self.department is None or self.department == line.department)
            and (self.location_key is None or self.location_key == line.location_key)
        )


@dataclass(frozen=True)
class ExpenseLine:
    line_id: str
    category: str
    role_level: str
    department: str
    location_key: str
    amount: Decimal
    submission_date: str
    trace_id: str = ""


@dataclass(frozen=True)
class CapDecision:
    """The verdict for one line, with a replayable audit trail."""
    line_id: str
    trace_id: str
    policy_version: str
    binding_cap_id: str
    binding_tier: str
    effective_limit: Decimal
    submitted_amount: Decimal
    in_policy: bool
    resolution_path: list[str]
    decision_hash: str
    evaluated_at: str

    def as_audit_event(self) -> dict:
        return {
            "line_id": self.line_id,
            "trace_id": self.trace_id,
            "policy_version": self.policy_version,
            "binding_cap_id": self.binding_cap_id,
            "binding_tier": self.binding_tier,
            "effective_limit": str(self.effective_limit),
            "submitted_amount": str(self.submitted_amount),
            "violation_code": None if self.in_policy else "CAP_EXCEEDED",
            "resolution_path": self.resolution_path,
            "decision_hash": self.decision_hash,
            "evaluated_at": self.evaluated_at,
        }


def snapshot_version(caps: Iterable[Cap]) -> str:
    """Deterministic SHA-256 over a canonically ordered cap set."""
    payload = sorted(
        (c.cap_id, int(c.tier), str(c.limit), c.category, c.role_level,
         c.department, c.location_key)
        for c in caps
    )
    return hashlib.sha256(
        json.dumps(payload, default=str, sort_keys=True).encode("utf-8")
    ).hexdigest()


class CapHierarchy:
    """Resolves the binding ceiling for a line against an immutable snapshot."""

    _CENTS = Decimal("0.01")

    def __init__(self, caps: list[Cap]):
        # Freeze the snapshot once; version-hash it so the whole batch audits
        # against exactly these rules regardless of later policy edits.
        self._caps: tuple[Cap, ...] = tuple(caps)
        self.policy_version: str = snapshot_version(self._caps)
        # Index by category (including the None wildcard) for bounded scanning.
        self._by_category: dict[Optional[str], list[Cap]] = {}
        for cap in self._caps:
            self._by_category.setdefault(cap.category, []).append(cap)

    def _candidates(self, line: ExpenseLine) -> list[Cap]:
        return self._by_category.get(line.category, []) + self._by_category.get(None, [])

    def resolve(self, line: ExpenseLine) -> CapDecision:
        applicable = [c for c in self._candidates(line) if c.applies_to(line)]
        if not applicable:
            raise ValueError(
                f"No applicable cap for line {line.line_id}; an ABSOLUTE "
                f"corporate cap must always match."
            )
        # Binding cap = smallest limit; ties broken by hardest tier (lowest ordinal).
        binding = min(applicable, key=lambda c: (c.limit, int(c.tier)))
        limit = binding.limit.quantize(self._CENTS, rounding=ROUND_HALF_UP)
        amount = line.amount.quantize(self._CENTS, rounding=ROUND_HALF_UP)
        in_policy = amount <= limit
        # Ordered path of every applicable cap = the audit-facing "why".
        path = [
            f"{c.tier.name}:{c.cap_id}={c.limit}"
            for c in sorted(applicable, key=lambda c: (int(c.tier), c.cap_id))
        ]
        decision_hash = hashlib.sha256(
            "|".join([self.policy_version, line.line_id, str(amount),
                      binding.cap_id, str(limit)]).encode("utf-8")
        ).hexdigest()
        return CapDecision(
            line_id=line.line_id,
            trace_id=line.trace_id or line.line_id,
            policy_version=self.policy_version,
            binding_cap_id=binding.cap_id,
            binding_tier=binding.tier.name,
            effective_limit=limit,
            submitted_amount=amount,
            in_policy=in_policy,
            resolution_path=path,
            decision_hash=decision_hash,
            evaluated_at=datetime.now(timezone.utc).isoformat(),
        )

    def validate_chunk(self, chunk: list[ExpenseLine]) -> list[CapDecision]:
        """Stateless per-chunk evaluation; every decision is audit-logged."""
        decisions = []
        for line in chunk:
            decision = self.resolve(line)
            audit_logger.info(json.dumps(decision.as_audit_event()))
            decisions.append(decision)
        return decisions


def stream_lines(source: Iterator[dict], chunk_size: int = 10_000
                 ) -> Iterator[list[ExpenseLine]]:
    """Generator-based chunking so memory stays flat across the whole ledger."""
    chunk: list[ExpenseLine] = []
    for raw in source:
        chunk.append(ExpenseLine(
            line_id=raw["id"],
            category=raw["category"],
            role_level=raw["role"],
            department=raw["dept"],
            location_key=raw["location"],
            amount=Decimal(str(raw["amount"])),
            submission_date=raw["date"],
            trace_id=raw.get("trace_id", ""),
        ))
        if len(chunk) >= chunk_size:
            yield chunk
            chunk = []
    if chunk:
        yield chunk


if __name__ == "__main__":
    # Immutable snapshot loaded once per run.
    caps = [
        Cap("CORP_ABS", CapTier.ABSOLUTE, Decimal("1000.00")),
        Cap("DEPT_SALES", CapTier.DEPARTMENT, Decimal("600.00"), department="SALES"),
        Cap("ROLE_DIR", CapTier.ROLE, Decimal("500.00"), role_level="DIRECTOR"),
        Cap("CAT_LODGING", CapTier.CATEGORY, Decimal("300.00"), category="LODGING"),
        Cap("PD_ZURICH_LODGING", CapTier.PER_DIEM, Decimal("380.00"),
            category="LODGING", location_key="ZRH"),
    ]
    hierarchy = CapHierarchy(caps)

    def mock_source() -> Iterator[dict]:
        for i in range(25_000):
            yield {"id": f"EXP-{i}", "category": "LODGING", "role": "DIRECTOR",
                   "dept": "SALES", "location": "ZRH",
                   "amount": "340.00" if i % 4 == 0 else "260.00",
                   "date": "2026-06-15", "trace_id": f"run-2026-06/{i}"}

    breaches = 0
    for batch in stream_lines(mock_source(), chunk_size=10_000):
        for decision in hierarchy.validate_chunk(batch):
            breaches += 0 if decision.in_policy else 1
    print(f"policy_version={hierarchy.policy_version[:12]} breaches={breaches}")

The binding cap for the worked example resolves to CAT_LODGING at 300.00 — the smallest applicable limit — so a 340.00 line is flagged CAP_EXCEEDED even though the Zurich per-diem override would allow 380.00, because a harder category ceiling still binds. That intersection logic is exactly what naive single-cap checks get wrong. The deeper precedence-compilation walkthrough is in implementing tiered spending caps in Python.

Configuration Reference

All tunable settings are explicit and version-pinned. Pin the resolver behind a config object rather than scattering literals through the loop, so the policy_version hash changes whenever any value changes.

Version-pinning guidance: treat the cap snapshot as an immutable artifact. Bump a semantic label (e.g. v2026.07_travel) in the source data, let snapshot_version recompute the hash, and store both alongside the run so an auditor can reconstruct the exact rule set. Dynamic, feedback-driven limits belong in Dynamic Threshold Tuning, which emits new snapshots this resolver consumes read-only.

Validation & Testing

Cap resolution is pure, so it tests cleanly with fixtures. Assert on the binding tier and effective limit, not just pass/fail, because a correct verdict for the wrong reason silently breaks the audit trail.

from decimal import Decimal

def _line(**kw) -> ExpenseLine:
    base = dict(line_id="T1", category="LODGING", role_level="DIRECTOR",
                department="SALES", location_key="ZRH",
                amount=Decimal("340.00"), submission_date="2026-06-15")
    base.update(kw)
    return ExpenseLine(**base)


def test_tightest_cap_binds_over_softer_override():
    h = CapHierarchy([
        Cap("CORP_ABS", CapTier.ABSOLUTE, Decimal("1000.00")),
        Cap("CAT_LODGING", CapTier.CATEGORY, Decimal("300.00"), category="LODGING"),
        Cap("PD_ZRH", CapTier.PER_DIEM, Decimal("380.00"),
            category="LODGING", location_key="ZRH"),
    ])
    d = h.resolve(_line())
    assert d.binding_cap_id == "CAT_LODGING"
    assert d.effective_limit == Decimal("300.00")
    assert d.in_policy is False


def test_equal_limits_break_to_hardest_tier():
    h = CapHierarchy([
        Cap("CORP_ABS", CapTier.ABSOLUTE, Decimal("300.00")),
        Cap("CAT_LODGING", CapTier.CATEGORY, Decimal("300.00"), category="LODGING"),
    ])
    d = h.resolve(_line(amount=Decimal("250.00")))
    assert d.binding_tier == "ABSOLUTE"   # ordinal 1 beats CATEGORY on a tie
    assert d.in_policy is True


def test_boundary_amount_is_in_policy():
    h = CapHierarchy([Cap("CORP_ABS", CapTier.ABSOLUTE, Decimal("300.00"))])
    assert h.resolve(_line(amount=Decimal("300.00"))).in_policy is True


def test_missing_cap_fails_closed():
    h = CapHierarchy([Cap("CAT_MEALS", CapTier.CATEGORY, Decimal("75.00"),
                          category="MEALS")])
    try:
        h.resolve(_line(category="LODGING"))
        assert False, "expected fail-closed ValueError"
    except ValueError:
        pass


def test_snapshot_hash_is_stable_and_order_independent():
    a = CapHierarchy([Cap("X", CapTier.ABSOLUTE, Decimal("10")),
                      Cap("Y", CapTier.ROLE, Decimal("5"), role_level="STAFF")])
    b = CapHierarchy([Cap("Y", CapTier.ROLE, Decimal("5"), role_level="STAFF"),
                      Cap("X", CapTier.ABSOLUTE, Decimal("10"))])
    assert a.policy_version == b.policy_version

Edge-case fixtures worth codifying: a boundary amount equal to the cap (must be in-policy, not a breach), a per-diem override that is softer than a category ceiling (the category must still bind), currency lines already normalized to a single settlement currency by the Per Diem Rate Structuring layer, and a split transaction whose two halves each clear a cap the whole would breach — deduplication of that pattern belongs to Merchant Category Code Routing and the anomaly layer, not here.

Operational Runbook

Deploy, monitor, and roll back the resolver as an immutable, snapshot-driven job:

1. Freeze the snapshot. Build the cap set from the policy repository, compute snapshot_version, and record the hash and its semantic label with the run manifest before any line is read.
2. Dry-run against a sampled window. Resolve a recent day’s ledger and diff the binding-tier distribution against the prior snapshot; an unexpected shift in which tier binds most often signals a mis-scoped cap.
3. Deploy read-only. The job consumes the snapshot and the categorized stream; it writes only violations and the append-only audit log. It never mutates policy data.
4. Monitor these signals:
   • Breach rate per category — a sudden spike usually means an upstream categorization or per-diem regression, not a real policy change.
   • Fail-closed ValueError count — must be zero in steady state; any occurrence means a line reached the resolver with no matching absolute cap.
   • Per-chunk latency variance — flat memory should keep p99 stable; drift indicates a chunk is materializing more than expected.
5. Alert thresholds. Page on any fail-closed error, on breach rate deviating more than ~3× the trailing daily median, and on chunk latency p99 exceeding twice the baseline.
6. Roll back by snapshot, not by code. Because verdicts are pinned to policy_version, reverting is re-running the prior snapshot hash — no code deploy required. Retain superseded snapshots so historical reports always audit against the rules active at submission time.

Verdicts leave this component and flow downstream into Automated Policy Validation & Anomaly Flagging, which correlates individual breaches with cross-line patterns.

Related

• Core Policy Architecture & Taxonomy Design
• Implementing tiered spending caps in Python
• Expense Category Taxonomies
• Per Diem Rate Structuring
• Security & Compliance Boundaries
• Dynamic Threshold Tuning