Quarantining Invalid CAM Expense Records
A record that fails validation during a CAM close is not garbage to discard — it is a vendor invoice with real dollars attached that must be held, explained, fixed, and re-run without ever silently vanishing or double-posting. This page designs the quarantine queue that holds those rejects: the typed error taxonomy that says why each record failed, the context envelope that captures enough to fix it without re-reading the source PDF, and the remediate-and-replay loop that returns a corrected record to the allocatable pool exactly once. It is the reject-path implementation behind schema validation for parsed expense data, the validation stage of the broader automated invoice parsing and data ingestion pipeline. Where the validator decides accept-or-reject, this queue owns everything that happens to a record after the reject — and getting that lifecycle wrong is how a legitimate $8,400.00 repair invoice quietly disappears three weeks before year-end tie-out.
Context & When to Use This Approach
A quarantine queue earns its keep the moment your feed stops being small enough to fix by hand. If a nightly ingest of a dozen trusted invoices throws a validation error, an accountant can open the PDF and re-key it. That does not survive a month-end close where three properties dump several thousand invoices from vendor portals, an accounts-payable inbox, and an SFTP drop within the same hour. At that volume the reject path needs to be a first-class system, not an exception log, and the signals that you have crossed the line are concrete:
- A single malformed record must never abort the surrounding batch — the other invoices in the same run have to keep flowing while the bad one is set aside.
- Accountants need to work rejects from the reject itself, with the exact failing rule, field, source coordinates, and document hash in front of them, instead of re-hunting the originating PDF for every failure.
- A rejected record is frequently fixable and legitimate — a transposed total, a credit memo parsed with the wrong sign, an OCR-mangled amount — so the queue must support correcting it and re-running it, not just deleting it.
- The same corrected invoice must re-enter the pipeline exactly once, even if a replay is retried, so a fixed $8,400.00 repair never posts twice against a tenant’s pro-rata share.
This is the sibling concern to building a CAM data validation layer: that page builds the gate that decides accept-or-quarantine and streams a batch through it; this page owns what the queue does with the rejects afterward. The two share a philosophy with retrying failed invoice batches with idempotency — a poison record is held rather than allowed to sink a run, and every re-run is keyed so it cannot double-count. If you are only validating a handful of trusted invoices interactively, skip all of this and let the model raise. The lifecycle below is for the case where a reject has to be captured, explained, fixed, and replayed under audit.
Assume the field-level models already exist. The ParsedCAMInvoice Pydantic model and the to_cents money coercion helper are defined in the parent schema validation for parsed expense data; this page imports them and builds the quarantine lifecycle around them. Every monetary value stays in Python’s decimal module — never float — so that summing thousands of quarantined-then-remediated amounts never drifts a reconciliation past its one-cent tolerance.
Step-by-Step Implementation
The lifecycle is five stages: validate and classify the failure by type, wrap the reject in a context envelope, route it to a keyed quarantine store, remediate and replay it exactly once, and instrument the queue so a spike is visible before it becomes a tie-out surprise.
Step 1 — Validate with Pydantic and classify every failure by type
Raw exceptions are hostile to a queue: “1 validation error” tells an accountant nothing sortable. Convert every failure into a typed taxonomy so the queue can group rejects by cause, route them to the right person, and drive metrics. Map Pydantic’s structured error output onto a small closed set of categories that mean something in CAM terms.
from __future__ import annotations
import enum
from dataclasses import dataclass
from decimal import Decimal
from typing import Optional
from pydantic import ValidationError as PydanticValidationError
# Field-level model + money helper live in the parent stage's implementation.
from cam.schema import ParsedCAMInvoice
class ErrorCategory(str, enum.Enum):
"""Closed taxonomy of why a CAM record fails — sortable in the queue."""
SCHEMA_TYPE = "schema_type" # wrong type / uncoercible value
MISSING_FIELD = "missing_field" # a required field is absent
RECONCILIATION = "reconciliation" # line items do not sum to the total
OUT_OF_RANGE = "out_of_range" # amount, date, or share outside bounds
SIGN_CONTRACT = "sign_contract" # credit/reversal with unexpected sign
PROVENANCE = "provenance" # missing or mismatched source hash
@dataclass(frozen=True)
class TypedError:
"""One machine-readable, categorized reason a record failed."""
category: ErrorCategory
field: Optional[str] # dotted path, e.g. "line_items.0.amount"
message: str
def _classify(err: dict) -> ErrorCategory:
"""Map one Pydantic error dict onto the CAM taxonomy."""
kind = err.get("type", "")
if kind in {"missing", "model_attributes_type"}:
return ErrorCategory.MISSING_FIELD
if kind.endswith("_parsing") or kind.endswith("_type"):
return ErrorCategory.SCHEMA_TYPE
if "greater_than" in kind or "less_than" in kind:
return ErrorCategory.OUT_OF_RANGE
return ErrorCategory.SCHEMA_TYPE
def classify_failure(raw: dict) -> list[TypedError]:
"""Validate a raw record; return [] if clean, else typed errors."""
try:
ParsedCAMInvoice.model_validate(raw)
except PydanticValidationError as exc:
return [
TypedError(
category=_classify(err),
field=".".join(str(p) for p in err["loc"]) or None,
message=err["msg"],
)
for err in exc.errors()
]
return []The taxonomy is deliberately closed. A new failure mode should force a conscious decision to add a category, not silently land in a catch-all — because “unknown error” records are exactly the ones that rot in a queue nobody triages.
Step 2 — Capture full context into a quarantine record
A reject that carries only “reconciliation failed” makes an accountant reopen the PDF. Capture enough context that the fix can happen from the queue entry alone: the original raw payload (so it can be patched and replayed), the source document hash (so provenance survives), the typed errors, an attempt counter, and lifecycle timestamps. The source_hash is bound to the immutable stored document, never to the parse, so it is stable across every replay.
import hashlib
import json
from datetime import datetime, timezone
class QuarantineStatus(str, enum.Enum):
NEW = "new" # just quarantined, awaiting triage
IN_REVIEW = "in_review" # an accountant has claimed it
REPLAYED = "replayed" # a remediation was re-run
RESOLVED = "resolved" # replay passed; record is in the pool
ABANDONED = "abandoned" # duplicate or unfixable; closed with reason
def _now() -> datetime:
return datetime.now(timezone.utc)
@dataclass
class QuarantineRecord:
"""A held reject with everything needed to fix and replay it."""
quarantine_id: str
invoice_id: str
source_hash: str
dominant_category: ErrorCategory
errors: list[TypedError]
raw_payload: dict # the exact parsed input, for patch + replay
status: QuarantineStatus
attempts: int
first_seen: datetime
last_updated: datetime
def build_quarantine(raw: dict, source_hash: str, errors: list[TypedError]) -> QuarantineRecord:
"""Wrap a failed record and its typed errors into a queue entry."""
invoice_id = str(raw.get("invoice_id", "UNKNOWN"))
# A deterministic id keyed on source + payload so re-quarantining the
# same reject collapses onto one entry rather than fanning out duplicates.
fingerprint = hashlib.sha256(
(source_hash + json.dumps(raw, sort_keys=True)).encode("utf-8")
).hexdigest()[:16]
now = _now()
return QuarantineRecord(
quarantine_id=f"Q-{fingerprint}",
invoice_id=invoice_id,
source_hash=source_hash,
dominant_category=errors[0].category, # first error drives routing
errors=errors,
raw_payload=raw,
status=QuarantineStatus.NEW,
attempts=0,
first_seen=now,
last_updated=now,
)The quarantine_id is a content fingerprint, not a random UUID. That single decision is what makes the queue idempotent: re-ingesting the same broken invoice — a common outcome when an upstream retry replays a whole batch — updates the existing entry instead of spawning a second copy an accountant would have to reconcile.
Step 3 — Route failures to a quarantine queue keyed by content hash
The store is a keyed collection: upsert by quarantine_id, so the same reject seen twice never duplicates. In production this is a database table or a dead-letter queue; the interface below is what the rest of the lifecycle depends on. Keeping the key as the content fingerprint means an at-least-once ingest produces an exactly-once quarantine entry — the same guarantee that retrying failed invoice batches with idempotency applies to the happy path.
class QuarantineStore:
"""Idempotent, keyed store for held rejects (in-memory reference impl)."""
def __init__(self) -> None:
self._records: dict[str, QuarantineRecord] = {}
def upsert(self, record: QuarantineRecord) -> QuarantineRecord:
"""Insert a new reject or refresh an existing one — never duplicate."""
existing = self._records.get(record.quarantine_id)
if existing is not None:
# Same reject seen again: keep the original first_seen, refresh cause.
existing.errors = record.errors
existing.dominant_category = record.dominant_category
existing.last_updated = _now()
return existing
self._records[record.quarantine_id] = record
return record
def open_records(self) -> list[QuarantineRecord]:
"""Everything still awaiting resolution, oldest first for aging."""
open_states = {QuarantineStatus.NEW, QuarantineStatus.IN_REVIEW,
QuarantineStatus.REPLAYED}
return sorted(
(r for r in self._records.values() if r.status in open_states),
key=lambda r: r.first_seen,
)
def get(self, quarantine_id: str) -> Optional[QuarantineRecord]:
return self._records.get(quarantine_id)Step 4 — Remediate and replay without double-counting
Remediation is a patch — a small mapping of field overrides an accountant supplies (or a rule proposes) — applied to the held raw_payload, after which the record is re-validated through the same gate from Step 1. If it now passes, it is promoted to RESOLVED and the certified invoice is returned for the allocatable pool; if it still fails, the attempt is recorded and it stays open with fresh errors. Because promotion is guarded by status, replaying an already-RESOLVED record is a no-op — the corrected invoice cannot post twice.
@dataclass(frozen=True)
class ReplayResult:
quarantine_id: str
resolved: bool
invoice: Optional[ParsedCAMInvoice] # present only when resolved
def remediate_and_replay(
store: QuarantineStore,
quarantine_id: str,
patch: dict,
) -> ReplayResult:
"""Apply a field patch to a held reject and re-run validation once."""
record = store.get(quarantine_id)
if record is None:
raise KeyError(f"no quarantined record {quarantine_id}")
# Idempotency guard: an already-resolved record never re-posts.
if record.status is QuarantineStatus.RESOLVED:
return ReplayResult(quarantine_id, resolved=True, invoice=None)
# Apply the correction to a copy — the original raw_payload is preserved
# as the audit trail of what was actually parsed from the vendor PDF.
patched = {**record.raw_payload, **patch}
errors = classify_failure(patched)
record.attempts += 1
record.last_updated = _now()
if errors:
# Still broken: refresh the cause and keep it open for another pass.
record.errors = errors
record.dominant_category = errors[0].category
record.status = QuarantineStatus.REPLAYED
return ReplayResult(quarantine_id, resolved=False, invoice=None)
# Clean now: promote and hand the certified invoice back to the pipeline.
record.raw_payload = patched
record.status = QuarantineStatus.RESOLVED
invoice = ParsedCAMInvoice.model_validate(patched)
return ReplayResult(quarantine_id, resolved=True, invoice=invoice)Note that the patch is applied to a copy and the promotion is idempotent. Between those two properties, remediation is safe to retry, safe to run concurrently against distinct records, and always leaves an audit trail of the original parse next to the correction.
Step 5 — Instrument the queue: aging, category mix, and replay success
A quarantine queue that nobody watches is a silent backlog that surfaces as a blown year-end tie-out. Emit three numbers continuously: how the open rejects break down by category (which vendor template just changed), how old the oldest open reject is (are we falling behind the close deadline), and what share of remediations actually stick. The replay success rate is worth a formula, where and are terminal counts:
from collections import Counter
@dataclass(frozen=True)
class QueueMetrics:
open_count: int
by_category: dict[str, int]
oldest_age_seconds: Decimal
replay_success: Decimal # resolved / (resolved + abandoned)
def queue_metrics(store: QuarantineStore) -> QueueMetrics:
"""Snapshot the queue health for a dashboard or alert threshold."""
open_records = store.open_records()
categories = Counter(r.dominant_category.value for r in open_records)
if open_records:
age = _now() - open_records[0].first_seen
oldest = Decimal(str(age.total_seconds())).quantize(Decimal("1"))
else:
oldest = Decimal("0")
all_records = list(store._records.values())
resolved = sum(1 for r in all_records if r.status is QuarantineStatus.RESOLVED)
abandoned = sum(1 for r in all_records if r.status is QuarantineStatus.ABANDONED)
terminal = resolved + abandoned
success = (
(Decimal(resolved) / Decimal(terminal)).quantize(Decimal("0.01"))
if terminal else Decimal("0")
)
return QueueMetrics(
open_count=len(open_records),
by_category=dict(categories),
oldest_age_seconds=oldest,
replay_success=success,
)A sudden concentration of reconciliation rejects against one vendor almost always means that vendor changed its statement template — a signal far cheaper to catch on this dashboard than in a tenant dispute months later.
Gotchas & Known Limitations
The failure modes that actually bite are in the lifecycle, not the field rules. Treat this as a pre-flight checklist before trusting the queue against a live feed:
Verification
Confirm the lifecycle by asserting on each stage independently: a clean record classifies to no errors, a defect classifies to the specific category that should fire, the store collapses duplicates, and a replay resolves a fixable reject exactly once while leaving an already-resolved record untouched. Assert on the taxonomy and the idempotency guard so a later refactor cannot quietly weaken either.
from decimal import Decimal
def _clean_raw() -> dict:
return {
"invoice_id": "INV-7781",
"vendor_name": "Beacon Landscaping",
"invoice_date": "2026-03-11",
"property_id": "PLAZA-9",
"total_amount": "2450.00",
"line_items": [
{"description": "Grounds", "gl_code": "5200-LAND", "amount": "2450.00"},
],
}
def test_clean_record_has_no_typed_errors() -> None:
assert classify_failure(_clean_raw()) == []
def test_missing_field_classifies_precisely() -> None:
raw = _clean_raw()
del raw["total_amount"]
errors = classify_failure(raw)
assert any(e.category is ErrorCategory.MISSING_FIELD for e in errors)
def test_store_collapses_duplicate_rejects() -> None:
store = QuarantineStore()
raw = _clean_raw()
del raw["property_id"]
errors = classify_failure(raw)
first = store.upsert(build_quarantine(raw, "sha256:aa", errors))
again = store.upsert(build_quarantine(raw, "sha256:aa", errors))
assert first.quarantine_id == again.quarantine_id # same content key
assert len(store._records) == 1 # no duplicate entry
def test_remediation_resolves_once_and_is_idempotent() -> None:
store = QuarantineStore()
raw = _clean_raw()
raw["total_amount"] = "24500.00" # a stray zero: fails reconciliation intent
errors = classify_failure(raw) or [
TypedError(ErrorCategory.RECONCILIATION, "total_amount", "does not tie")
]
rec = store.upsert(build_quarantine(raw, "sha256:bb", errors))
result = remediate_and_replay(store, rec.quarantine_id, {"total_amount": "2450.00"})
assert result.resolved is True
assert store.get(rec.quarantine_id).status is QuarantineStatus.RESOLVED
assert store.get(rec.quarantine_id).attempts == 1
# Replaying a resolved record must not re-post or re-increment.
repeat = remediate_and_replay(store, rec.quarantine_id, {"total_amount": "2450.00"})
assert repeat.resolved is True
assert store.get(rec.quarantine_id).attempts == 1 # idempotent no-opThe last test is the one that matters most: it proves the corrected invoice resolves once and that a retried replay is a genuine no-op, so a fixed amount can never post twice against a tenant’s share. Extend it with property-based tests that generate reconciling and non-reconciling line-item sets in Decimal, assert the reconciling ones classify clean, and assert every non-reconciling one lands in quarantine with a category an accountant can route.
Once the queue is green, resolved records rejoin the certified stream that flows to GL code mapping for CAM expenses, and the whole design belongs to schema validation for parsed expense data; the gate that feeds this queue is built in building a CAM data validation layer.
Related
- Schema validation for parsed expense data — the parent stage whose Pydantic models decide the accept-or-quarantine outcome this queue then owns.
- Building a CAM data validation layer — the sibling gate that streams a batch and routes each reject into this queue.
- Retrying failed invoice batches with idempotency — the same exactly-once, content-keyed discipline applied to the batch-retry path.