"""Hard and soft validation rules over a parsed ``Document`` (pure, no I/O). Validation is the precision lever for the auto-accept path (see CLAUDE.md "Precision posture"). Two classes of rule run over a parsed ``Document`` and produce a structured ``ValidationReport``: - **Hard rules (H1-H4):** a failure forces ``review`` regardless of model confidence. These are the arithmetic cross-checks and critical-field presence/type guards -- the mechanism that catches a confidently-wrong number before it is written. - **Soft rules (S1-S4):** a failure reduces confidence but does not by itself force review. They surface "looks off" signals (missing vendor, implausible date, unknown currency, per-line arithmetic drift). A rule whose inputs are absent is **skipped** (status ``"skip"``), not failed: an absent subtotal must not spuriously fail the reconciliation check and push a valid document to review (that would cost recall for no precision gain). The one deliberate exception is ``H4`` -- an absent ``total`` is a hard failure, because a document with no total is never safe to auto-accept. Every function here is pure: no file, network, clock, or database access. The ``S1`` future-date check takes an injected ``today`` reference so the rule stays deterministic and unit-testable; the core passes ``date.today()`` in production. See ``docs/03_data_and_extraction_spec.md`` section 3 for the rule definitions and the monetary-epsilon policy. """ from __future__ import annotations from dataclasses import dataclass from datetime import date, timedelta from typing import Any, Literal from doc_agent.schema.models import Document, LineItem # --- Monetary comparison policy ------------------------------------------------- # Epsilon for monetary reconciliation accommodates rounding. Per the data spec, # the tolerance is the larger of an absolute floor and a small relative term, so # small receipts are compared to the cent while large invoices tolerate the # accumulated rounding of many line items. MONETARY_ABS_EPSILON: float = 0.02 MONETARY_REL_EPSILON: float = 0.005 # How many days a ``document_date`` may sit ahead of the reference "today" # before ``S1`` considers it implausibly future-dated (absorbs timezone skew). FUTURE_DATE_GRACE_DAYS: int = 1 # Critical fields, precision-prioritised (see CLAUDE.md). ``H1`` type-guards # these; missing/zero among them drives routing elsewhere. CRITICAL_FIELDS: tuple[str, ...] = ("total", "tax", "invoice_number") # Known ISO 4217 codes for the soft currency check. Intentionally a common # subset weighted toward the evaluation datasets (SROIE/CORD/MC-OCR cover # Singapore, Indonesia, Vietnam); an unrecognized-but-valid rare code only # incurs a small soft penalty, which is the precision-safe direction. KNOWN_CURRENCIES: frozenset[str] = frozenset( { "USD", "EUR", "GBP", "JPY", "CHF", "CAD", "AUD", "NZD", "CNY", "HKD", "SGD", "MYR", "IDR", "THB", "VND", "PHP", "INR", "KRW", "TWD", "MOP", "SEK", "NOK", "DKK", "PLN", "CZK", "HUF", "RON", "RUB", "TRY", "UAH", "ZAR", "BRL", "MXN", "ARS", "CLP", "COP", "AED", "SAR", "QAR", "ILS", "EGP", "NGN", "KES", "PKR", "BDT", "LKR", } ) RuleSeverity = Literal["hard", "soft"] RuleStatus = Literal["pass", "fail", "skip"] def money_close( left: float, right: float, *, abs_epsilon: float = MONETARY_ABS_EPSILON, rel_epsilon: float = MONETARY_REL_EPSILON, ) -> bool: """Compare two monetary amounts within the rounding tolerance. The tolerance is ``max(abs_epsilon, rel_epsilon * max(|left|, |right|))`` -- the larger of an absolute floor and a small relative term (data spec section 3). Args: left: First amount. right: Second amount. abs_epsilon: Absolute tolerance floor. Defaults to ``MONETARY_ABS_EPSILON``. rel_epsilon: Relative tolerance fraction. Defaults to ``MONETARY_REL_EPSILON``. Returns: ``True`` if the amounts are equal within tolerance. """ tolerance = max(abs_epsilon, rel_epsilon * max(abs(left), abs(right))) return abs(left - right) <= tolerance @dataclass(frozen=True) class RuleResult: """Outcome of a single validation rule. Attributes: code: Rule identifier ("H1"-"H4", "S1"-"S4"). severity: "hard" (a failure forces review) or "soft" (reduces score). status: "pass", "fail", or "skip" (inputs absent / not applicable). message: Short human-readable explanation of the outcome. """ code: str severity: RuleSeverity status: RuleStatus message: str def to_dict(self) -> dict[str, str]: """Serialize to a plain JSON-friendly dict. Returns: A dict with ``code``, ``severity``, ``status``, and ``message``. """ return { "code": self.code, "severity": self.severity, "status": self.status, "message": self.message, } @dataclass(frozen=True) class ValidationReport: """Structured result of running every rule over one ``Document``. The report is pure data: routing consumes ``hard_failed`` to short-circuit to review and ``soft_failures`` to penalize the confidence score. It is JSON-serializable via ``to_dict`` for storage in ``Document.validation``. Attributes: results: One ``RuleResult`` per rule, in rule order. """ results: tuple[RuleResult, ...] @property def hard_failures(self) -> tuple[RuleResult, ...]: """The hard rules that failed (empty if none).""" return tuple( r for r in self.results if r.severity == "hard" and r.status == "fail" ) @property def soft_failures(self) -> tuple[RuleResult, ...]: """The soft rules that failed (empty if none).""" return tuple( r for r in self.results if r.severity == "soft" and r.status == "fail" ) @property def hard_failed(self) -> bool: """Whether any hard rule failed (forces ``review`` downstream).""" return bool(self.hard_failures) def by_code(self, code: str) -> RuleResult | None: """Return the result for a rule code, or ``None`` if absent. Args: code: A rule identifier such as "H2". Returns: The matching ``RuleResult``, or ``None``. """ for result in self.results: if result.code == code: return result return None def to_dict(self) -> dict[str, Any]: """Serialize the report for storage in ``Document.validation``. Returns: A dict with ``hard_failed`` flag, the full ``results`` list, and the codes of the hard and soft failures for quick inspection. """ return { "hard_failed": self.hard_failed, "results": [r.to_dict() for r in self.results], "hard_failures": [r.code for r in self.hard_failures], "soft_failures": [r.code for r in self.soft_failures], } # --- Hard rules ----------------------------------------------------------------- def _check_h1_critical_types(document: Document) -> RuleResult: """H1: present critical fields hold the correct type. The schema already enforces types on construction, so this is a defensive contract guard: ``total``/``tax`` must be numeric and ``invoice_number`` a string when present. """ bad: list[str] = [] for name in ("total", "tax"): value = getattr(document, name) if value is not None and (isinstance(value, bool) or not isinstance(value, (int, float))): bad.append(name) if document.invoice_number is not None and not isinstance(document.invoice_number, str): bad.append("invoice_number") if bad: return RuleResult("H1", "hard", "fail", f"critical field(s) mistyped: {', '.join(bad)}") return RuleResult("H1", "hard", "pass", "critical fields are correctly typed") def _check_h2_totals_reconcile(document: Document) -> RuleResult: """H2: subtotal + tax approximately equals total, when all three exist.""" subtotal, tax, total = document.subtotal, document.tax, document.total if subtotal is None or tax is None or total is None: return RuleResult("H2", "hard", "skip", "subtotal, tax, or total absent") if money_close(subtotal + tax, total): return RuleResult("H2", "hard", "pass", f"{subtotal} + {tax} == {total}") return RuleResult( "H2", "hard", "fail", f"{subtotal} + {tax} != {total} (got {subtotal + tax})" ) def _sum_line_amounts(line_items: list[LineItem]) -> float | None: """Sum line-item amounts, or ``None`` if any amount is missing. Reconciliation is only meaningful when every term is present; a single missing amount makes the sum incomplete, so the check is skipped rather than run against an understated total. """ total = 0.0 for item in line_items: if item.amount is None: return None total += item.amount return total def _check_h3_line_items_reconcile(document: Document) -> RuleResult: """H3: sum(line_items.amount) approximately equals subtotal (or total).""" if not document.line_items: return RuleResult("H3", "hard", "skip", "no line items") line_sum = _sum_line_amounts(document.line_items) if line_sum is None: return RuleResult("H3", "hard", "skip", "one or more line items lack an amount") reference_name = "subtotal" if document.subtotal is not None else "total" reference = document.subtotal if document.subtotal is not None else document.total if reference is None: return RuleResult("H3", "hard", "skip", "no subtotal or total to reconcile against") if money_close(line_sum, reference): return RuleResult("H3", "hard", "pass", f"line sum {line_sum} == {reference_name} {reference}") return RuleResult( "H3", "hard", "fail", f"line sum {line_sum} != {reference_name} {reference}" ) def _check_h4_total_present(document: Document) -> RuleResult: """H4: total is present and non-negative.""" total = document.total if total is None: return RuleResult("H4", "hard", "fail", "total is missing") if total < 0: return RuleResult("H4", "hard", "fail", f"total is negative ({total})") return RuleResult("H4", "hard", "pass", f"total present and non-negative ({total})") # --- Soft rules ----------------------------------------------------------------- def _check_s1_date_plausible(document: Document, today: date | None) -> RuleResult: """S1: document_date is present and not implausibly far in the future. The future check only runs when a ``today`` reference is supplied (keeping the function pure); presence is always checked. """ if document.document_date is None: return RuleResult("S1", "soft", "fail", "document_date is missing") if today is not None: latest = today + timedelta(days=FUTURE_DATE_GRACE_DAYS) if document.document_date > latest: return RuleResult( "S1", "soft", "fail", f"document_date {document.document_date} is in the future" ) return RuleResult("S1", "soft", "pass", f"document_date {document.document_date} is plausible") def _check_s2_currency_known(document: Document) -> RuleResult: """S2: currency resolves to a known ISO 4217 code.""" currency = document.currency if currency is None: return RuleResult("S2", "soft", "fail", "currency is missing") if currency not in KNOWN_CURRENCIES: return RuleResult("S2", "soft", "fail", f"currency {currency!r} is not a known code") return RuleResult("S2", "soft", "pass", f"currency {currency} is a known code") def _check_s3_vendor_present(document: Document) -> RuleResult: """S3: vendor_name is non-empty (blank already normalized to ``None``).""" if document.vendor_name is None: return RuleResult("S3", "soft", "fail", "vendor_name is missing") return RuleResult("S3", "soft", "pass", "vendor_name is present") def _check_s4_line_arithmetic(document: Document) -> RuleResult: """S4: quantity * unit_price approximately equals amount, per line.""" checkable = 0 failures: list[int] = [] for index, item in enumerate(document.line_items): if item.quantity is None or item.unit_price is None or item.amount is None: continue checkable += 1 if not money_close(item.quantity * item.unit_price, item.amount): failures.append(index) if checkable == 0: return RuleResult("S4", "soft", "skip", "no line item has quantity, unit_price, and amount") if failures: rows = ", ".join(str(i) for i in failures) return RuleResult("S4", "soft", "fail", f"per-line arithmetic off at row(s): {rows}") return RuleResult("S4", "soft", "pass", "per-line arithmetic reconciles") def validate(document: Document, *, today: date | None = None) -> ValidationReport: """Run every hard and soft rule over a parsed document. Pure: no I/O. The ``today`` reference is injected (not read from the clock) so the ``S1`` future-date check stays deterministic; the core passes ``date.today()`` in production and tests pass a fixed date. Args: document: The parsed, schema-validated document to check. today: Reference date for the ``S1`` future-date plausibility check. If ``None``, only date *presence* is checked, not future-dating. Returns: A ``ValidationReport`` with one ``RuleResult`` per rule, in rule order. """ results = ( _check_h1_critical_types(document), _check_h2_totals_reconcile(document), _check_h3_line_items_reconcile(document), _check_h4_total_present(document), _check_s1_date_plausible(document, today), _check_s2_currency_known(document), _check_s3_vendor_present(document), _check_s4_line_arithmetic(document), ) return ValidationReport(results=results)