Upload 20 files

src/Invoice Validation Functions/.claude/settings.json

@@ -0,0 +1,28 @@
+{
+  "permissions": {
+    "allow": [
+      "Bash(python -c ' *)",
+      "Read(//c/Users/bhuva/Downloads/**)",
+      "Read(//c/Users/bhuva/**)",
+      "Bash(python -c \"import glob; print\\('\\\\n'.join\\(glob.glob\\(r'c:\\\\Users\\\\bhuva\\\\Downloads\\\\**\\\\*.xlsx', recursive=True\\)\\)\\)\")",
+      "Bash(python -c \"import glob; files = glob.glob\\(r'c:\\\\Users\\\\bhuva\\\\Downloads\\\\*master*.xlsx'\\); print\\('\\\\n'.join\\(files\\)\\)\")",
+      "Bash(python \"C:/Users/bhuva/AppData/Local/Temp/claude/c--Users-bhuva-Downloads-Invoice-Validation-Functions/52024fb1-90e8-497b-b1a9-726c0979fbcb/scratchpad/fix2.py\")",
+      "PowerShell(python \"C:\\\\Users\\\\bhuva\\\\AppData\\\\Local\\\\Temp\\\\claude\\\\c--Users-bhuva-Downloads-Invoice-Validation-Functions\\\\52024fb1-90e8-497b-b1a9-726c0979fbcb\\\\scratchpad\\\\audit_coverage.py\" 2>&1)",
+      "PowerShell(python \"C:\\\\Users\\\\bhuva\\\\AppData\\\\Local\\\\Temp\\\\claude\\\\c--Users-bhuva-Downloads-Invoice-Validation-Functions\\\\52024fb1-90e8-497b-b1a9-726c0979fbcb\\\\scratchpad\\\\verify_tc_flags.py\" 2>&1)",
+      "Bash(python \"C:\\\\Users\\\\bhuva\\\\AppData\\\\Local\\\\Temp\\\\claude\\\\c--Users-bhuva-Downloads-Invoice-Validation-Functions\\\\52024fb1-90e8-497b-b1a9-726c0979fbcb\\\\scratchpad\\\\verify_tc_flags.py\")",
+      "Bash(python -c \"import json; data=[json.loads\\(l\\) for l in open\\(r'C:/Users/bhuva/Downloads/test_invoices_comprehensive.jsonl'\\)]; [print\\(i+1, d.get\\('invoice_number'\\)\\) for i,d in enumerate\\(data\\)]\")",
+      "Bash(python -c \"import json; data=[json.loads\\(l\\) for l in open\\(r'C:/Users/bhuva/Downloads/test_invoices_comprehensive.jsonl'\\)]; print\\(list\\(data[7].keys\\(\\)\\)[:20]\\); print\\(json.dumps\\(data[7], indent=2, default=str\\)[:1000]\\)\")",
+      "Bash(python \"C:/Users/bhuva/AppData/Local/Temp/claude/c--Users-bhuva-Downloads-Invoice-Validation-Functions/52024fb1-90e8-497b-b1a9-726c0979fbcb/scratchpad/verify_tc_flags.py\")",
+      "Bash(powershell -Command \"Get-Content 'C:\\\\Users\\\\bhuva\\\\.claude\\\\projects\\\\c--Users-bhuva-Downloads-Invoice-Validation-Functions\\\\52024fb1-90e8-497b-b1a9-726c0979fbcb.jsonl' | Select-Object -Last 300\")",
+      "Read(//c/Users/bhuva/.claude/projects/c--Users-bhuva-Downloads-Invoice-Validation-Functions/**)",
+      "Bash(python3 -c ' *)",
+      "Bash(find /c/Users/bhuva -iname \"*comprehensive*\" 2>/dev/null | head -50)",
+      "Bash(grep -n \"TC019\\\\|TC-019\" \"/c/Users/bhuva/Downloads/test_invoices_comprehensive.jsonl\")",
+      "Bash(pip show *)"
+    ],
+    "additionalDirectories": [
+      "c:\\Users\\bhuva\\Downloads",
+      "C:\\Users\\bhuva\\Downloads"
+    ]
+  }
+}

src/Invoice Validation Functions/.claude/settings.local.json

@@ -0,0 +1,11 @@
+{
+  "permissions": {
+    "allow": [
+      "Bash(rm \"c:/Users/bhuva/Downloads/Matching Functions/pyrightconfig.json\")",
+      "PowerShell(cd \"c:\\\\Users\\\\bhuva\\\\Downloads\\\\Matching Functions\"; $env:PYTHONIOENCODING=\"utf-8\"; python edge_case_tests.py 2>&1)"
+    ],
+    "additionalDirectories": [
+      "c:\\Users\\bhuva\\Downloads\\Matching Functions"
+    ]
+  }
+}

src/Invoice Validation Functions/Bank validation.py

@@ -0,0 +1,449 @@
+import re
+from datetime import date, datetime
+from hashlib import sha256
+from unicodedata import normalize
+from decimal import Decimal
+
+from rapidfuzz import fuzz
+
+
+def run_bank_validation(
+    *,
+    document,
+    extracted_fields,
+    vendor_bank_accounts,
+    resolved_vendor_id,
+    platform_configs=None,
+    name_match_fn=None,
+):
+    """
+    Bank validation logic based on Cat 2 section 7.4.
+
+    Expected document fields:
+        document.id
+        document.tenant_id
+
+    Expected extracted_fields fields:
+        invoice_date
+        currency
+        bank_iban
+        bank_acc_no
+        bank_swift
+        bank_acc_name
+
+    Expected vendor_bank_accounts row fields:
+        tenant_id
+        vendor_id
+        iban_canonical
+        acc_no_canonical
+        swift_bic
+        bank_acc_name
+        currency_code
+        bank_country
+        is_primary
+        effective_from
+        effective_to
+        deleted_at
+
+    name_match_fn:
+        Optional function for bank account name comparison.
+        Should return something like:
+            {"decision": "matched" | "suggestion" | "unmatched"}
+    """
+
+    platform_configs = platform_configs or {}
+
+    iban_checksum_required = platform_configs.get(
+        "validation.iban_checksum_required",
+        True,
+    )
+
+    iban_accno_exempt_countries = set(
+        platform_configs.get(
+            "validation.iban_accno_substring_exempt_countries",
+            ["LC", "MT"],
+        )
+    )
+
+    high_threshold = Decimal(str(platform_configs.get("validation.rapidfuzz_high_threshold", "0.90")))
+    medium_threshold = Decimal(str(platform_configs.get("validation.rapidfuzz_medium_threshold", "0.75")))
+    suggestion_threshold = Decimal(str(platform_configs.get("validation.rapidfuzz_suggestion_threshold", "0.60")))
+
+    def hash_prefix(value, length=8):
+        if not value:
+            return None
+        return sha256(str(value).encode("utf-8")).hexdigest()[:length]
+
+    def canonicalize_text(value):
+        if value is None:
+            return None
+        value = normalize("NFKC", str(value))
+        value = value.strip().upper()
+        return value or None
+
+    def canonicalize_iban(value):
+        if value is None:
+            return None
+        value = normalize("NFKC", str(value))
+        value = re.sub(r"[\s\-\.,/\\:;_]+", "", value)
+        value = value.upper()
+        return value or None
+
+    def canonicalize_acc_no(value):
+        if value is None:
+            return None
+        value = normalize("NFKC", str(value))
+        value = re.sub(r"[\s\-\.,/\\:;_]+", "", value)
+        value = value.upper()
+        return value or None
+
+    def canonicalize_swift(value):
+        if value is None:
+            return None
+        value = normalize("NFKC", str(value))
+        value = re.sub(r"\s+", "", value)
+        value = value.upper()
+        return value or None
+
+    def iban_country(iban):
+        if not iban or len(iban) < 2:
+            return None
+        return iban[:2]
+
+    def iban_checksum_mod97_valid(iban):
+        """
+        Standard IBAN mod-97 validation.
+
+        Move first 4 chars to end.
+        Replace letters A=10 ... Z=35.
+        Result mod 97 must equal 1.
+        """
+        if not iban or len(iban) < 4:
+            return False
+
+        rearranged = iban[4:] + iban[:4]
+        numeric = ""
+
+        for char in rearranged:
+            if char.isdigit():
+                numeric += char
+            elif "A" <= char <= "Z":
+                numeric += str(ord(char) - ord("A") + 10)
+            else:
+                return False
+
+        remainder = 0
+        for digit in numeric:
+            remainder = (remainder * 10 + int(digit)) % 97
+
+        return remainder == 1
+
+    def iban_acc_no_consistent(iban, acc_no):
+        """
+        Basic consistency check:
+        remove country + check digits from IBAN,
+        then check whether account number appears inside BBAN.
+
+        Leading zeros are ignored for comparison.
+        """
+        if not iban or not acc_no:
+            return True
+
+        country = iban_country(iban)
+
+        if country in iban_accno_exempt_countries:
+            return True
+
+        bban = iban[4:]
+        stripped_acc_no = acc_no.lstrip("0") or acc_no
+
+        return stripped_acc_no in bban
+
+    def is_effective(row, invoice_date):
+        if getattr(row, "deleted_at", None):
+            return False
+
+        if invoice_date is None:
+            return False
+
+        effective_from = getattr(row, "effective_from", None)
+        effective_to = getattr(row, "effective_to", None)
+
+        if effective_from is not None and invoice_date < effective_from:
+            return False
+
+        if effective_to and invoice_date >= effective_to:
+            return False
+
+        return True
+
+    def default_name_match(invoice_name, master_name):
+        if not invoice_name or not master_name:
+            return {"decision": "unmatched"}
+
+        score = Decimal(str(fuzz.WRatio(invoice_name, master_name) / 100)).quantize(
+            Decimal("0.0001")
+        )
+
+        if score >= high_threshold:
+            return {"decision": "matched", "score_tier": "high"}
+        if score >= medium_threshold:
+            return {"decision": "matched", "score_tier": "medium"}
+        if score >= suggestion_threshold:
+            return {"decision": "suggestion", "score_tier": "low"}
+        return {"decision": "unmatched", "score_tier": "low"}
+
+    def bank_account_name_matches(invoice_name, master_name):
+        matcher = name_match_fn or default_name_match
+
+        result = matcher(
+            query_name=invoice_name,
+            candidate_pool=[master_name],
+            context={"purpose": "bank_acc_name_check"},
+        ) if name_match_fn else matcher(invoice_name, master_name)
+
+        return result.get("decision") == "matched"
+
+    def add_flag(code, *, field=None, weight=None, match_result="failed", invoice_value=None,
+                 master_value=None, currency_invoice=None, currency_master=None,
+                 primary_or_non_primary=None):
+        flag = {
+            "code": code,
+            "field": field,
+            "severity": "advisory",
+        }
+
+        if weight is not None:
+            flag["weight"] = weight
+
+        risk_flags.append(flag)
+
+        audit_entries.append(
+            {
+                "field": field,
+                "match_result": match_result,
+                "invoice_hash_prefix": hash_prefix(invoice_value),
+                "master_hash_prefix": hash_prefix(master_value),
+                "currency_invoice": currency_invoice,
+                "currency_master": currency_master,
+                "primary_or_non_primary": primary_or_non_primary,
+            }
+        )
+
+    risk_flags = []
+    advisory_flags = []
+    skipped_steps = []
+    audit_entries = []
+
+    if resolved_vendor_id is None:
+        skipped_steps.append(
+            {
+                "step": "bank_validation",
+                "skip_reason": "vendor_unresolved",
+            }
+        )
+
+        return {
+            "risk_flags": risk_flags,
+            "advisory_flags": advisory_flags,
+            "skipped_steps": skipped_steps,
+            "audit_entries": audit_entries,
+            "match_result": "skipped",
+        }
+
+    invoice_date = getattr(extracted_fields, "invoice_date", None)
+
+    if isinstance(invoice_date, datetime):
+        invoice_date = invoice_date.date()
+
+
+    invoice_iban = canonicalize_iban(getattr(extracted_fields, "bank_iban", None))
+    invoice_acc_no = canonicalize_acc_no(getattr(extracted_fields, "bank_acc_no", None))
+    invoice_swift = canonicalize_swift(getattr(extracted_fields, "bank_swift", None))
+    invoice_bank_acc_name = getattr(extracted_fields, "bank_acc_name", None)
+    invoice_currency = canonicalize_text(getattr(extracted_fields, "currency", None))
+
+    # 1. Invoice-side pre-format checks.
+    if invoice_iban and iban_checksum_required:
+        if not iban_checksum_mod97_valid(invoice_iban):
+            add_flag(
+                "bank_iban_invalid_checksum",
+                field="bank_iban",
+                weight=Decimal("0.35"),
+                invoice_value=invoice_iban,
+            )
+
+    if invoice_iban and invoice_acc_no:
+        if not iban_acc_no_consistent(invoice_iban, invoice_acc_no):
+            add_flag(
+                "bank_iban_accno_inconsistent",
+                field="bank_iban/bank_acc_no",
+                weight=Decimal("0.35"),
+                invoice_value=f"{invoice_iban}|{invoice_acc_no}",
+            )
+
+    # 2. Build same-tenant, same-vendor, active candidate pool.
+    # all_vendor_rows: non-deleted rows for this vendor, no date filtering.
+    # Used for new_bank_details_no_master — distinguishes "no rows at all" from
+    # "rows exist but none are currently effective".
+    all_vendor_rows = [
+        row
+        for row in vendor_bank_accounts
+        if getattr(row, "tenant_id", None) == document.tenant_id
+        and getattr(row, "vendor_id", None) == resolved_vendor_id
+        and not getattr(row, "deleted_at", None)
+    ]
+
+    candidate_pool = [
+        row for row in all_vendor_rows
+        if is_effective(row, invoice_date)
+    ]
+
+    candidate_pool.sort(
+        key=lambda row: (
+            bool(getattr(row, "is_primary", False)),
+            getattr(row, "effective_from", None) or date.max,
+        ),
+        reverse=True,
+    )
+
+    primary_rows = [
+        row for row in candidate_pool
+        if bool(getattr(row, "is_primary", False))
+    ]
+
+    non_primary_rows = [
+        row for row in candidate_pool
+        if not bool(getattr(row, "is_primary", False))
+    ]
+
+    primary = primary_rows[0] if primary_rows else None
+
+    invoice_has_any_bank_detail = any(
+        [
+            invoice_iban,
+            invoice_acc_no,
+            invoice_swift,
+            invoice_bank_acc_name,
+        ]
+    )
+
+    if invoice_has_any_bank_detail and not all_vendor_rows:
+        add_flag(
+            "new_bank_details_no_master",
+            field="bank_details",
+            invoice_value="bank_details_present",
+        )
+
+        return {
+            "risk_flags": risk_flags,
+            "advisory_flags": advisory_flags,
+            "skipped_steps": skipped_steps,
+            "audit_entries": audit_entries,
+            "match_result": "no_master_bank_details",
+        }
+
+    # 3. Compare against primary account.
+    if primary:
+        primary_match = True
+
+        master_iban = getattr(primary, "iban_canonical", None)
+        master_acc_no = getattr(primary, "acc_no_canonical", None)
+        master_swift = canonicalize_swift(getattr(primary, "swift_bic", None))
+        master_bank_acc_name = getattr(primary, "bank_acc_name", None)
+        master_currency = canonicalize_text(getattr(primary, "currency_code", None))
+
+        if invoice_iban:
+            if not master_iban:
+                primary_match = False
+                add_flag(
+                    "new_bank_field_type_no_master",
+                    field="bank_iban",
+                    invoice_value=invoice_iban,
+                    primary_or_non_primary="primary",
+                )
+            elif invoice_iban != master_iban:
+                primary_match = False
+                add_flag(
+                    "bank_iban_mismatch",
+                    field="bank_iban",
+                    invoice_value=invoice_iban,
+                    master_value=master_iban,
+                    primary_or_non_primary="primary",
+                )
+
+        if invoice_acc_no and master_acc_no and invoice_acc_no != master_acc_no:
+            primary_match = False
+            add_flag(
+                "bank_acc_no_mismatch",
+                field="bank_acc_no",
+                invoice_value=invoice_acc_no,
+                master_value=master_acc_no,
+                primary_or_non_primary="primary",
+            )
+
+        if invoice_swift and master_swift and invoice_swift != master_swift:
+            primary_match = False
+            add_flag(
+                "bank_swift_mismatch",
+                field="bank_swift",
+                invoice_value=invoice_swift,
+                master_value=master_swift,
+                primary_or_non_primary="primary",
+            )
+
+        if invoice_bank_acc_name and master_bank_acc_name:
+            if not bank_account_name_matches(invoice_bank_acc_name, master_bank_acc_name):
+                primary_match = False
+                add_flag(
+                    "bank_acc_name_mismatch",
+                    field="bank_acc_name",
+                    weight=Decimal("0.30"),
+                    invoice_value=invoice_bank_acc_name,
+                    master_value=master_bank_acc_name,
+                    primary_or_non_primary="primary",
+                )
+
+        if invoice_currency and master_currency and invoice_currency != master_currency:
+            primary_match = False
+            add_flag(
+                "bank_currency_mismatch",
+                field="currency",
+                weight=Decimal("0.20"),
+                currency_invoice=invoice_currency,
+                currency_master=master_currency,
+                primary_or_non_primary="primary",
+            )
+
+        match_result = "primary_match" if primary_match else "primary_mismatch"
+
+    else:
+        match_result = "no_primary_bank_account"
+
+        secondary_match = None
+        for row in non_primary_rows:
+            if invoice_iban and getattr(row, "iban_canonical", None) == invoice_iban:
+                secondary_match = row
+                break
+
+        if secondary_match:
+            add_flag(
+                "bank_non_primary_match",
+                field="bank_iban",
+                weight=Decimal("0.25"),
+                match_result="matched_non_primary",
+                invoice_value=invoice_iban,
+                master_value=getattr(secondary_match, "iban_canonical", None),
+                primary_or_non_primary="non_primary",
+            )
+
+    return {
+        "risk_flags": risk_flags,
+        "advisory_flags": advisory_flags,
+        "skipped_steps": skipped_steps,
+        "audit_entries": audit_entries,
+        "match_result": match_result,
+        "candidate_count": len(candidate_pool),
+        "primary_candidate_count": len(primary_rows),
+        "non_primary_candidate_count": len(non_primary_rows),
+    }

src/Invoice Validation Functions/CLAUDE.md.txt

@@ -0,0 +1,59 @@
+# Invoice Validation Project
+
+## Architecture
+
+This project implements an invoice validation engine.
+
+Validation modules are implemented separately and must remain separate.
+
+Current validation modules:
+
+1. Vendor Validation
+2. Tax ID Validation
+3. Bank Validation
+4. Date Sanity Check
+5. Payment Terms Validation
+6. Tax Validation
+7. Completeness Check
+
+## Rules
+
+* Do not merge validations into one large function.
+* Keep business logic inside each validation module.
+* The orchestration layer should only coordinate execution.
+* Vendor Validation runs first.
+* Any validation module that depends on vendor master data must receive resolved_vendor_id from Vendor Validation.
+* Pass resolved_vendor_id to all downstream validations that require vendor-specific lookups or checks.
+* All validation configuration is stored in platform_configs.json.
+* Validation functions receive platform_configs as input.
+* Do not hardcode operational thresholds.
+* Use the SDD as the source of truth.
+* Tenant isolation must always be respected.
+
+## Expected Outcome
+
+After all validation modules are verified against the SDD:
+
+* Create a lightweight orchestrator.
+* The orchestrator should load platform_configs.json once.
+* The orchestrator should call validators in the correct order.
+* The orchestrator should pass outputs from one validator to downstream validators where required.
+* The orchestrator should aggregate risk flags and advisory flags.
+* No business logic should be implemented in the orchestrator.
+
+
+## Matching Project Rules
+
+- Matching runs after validation.
+- Do not run all match types for every invoice.
+- First decide the match type using a routing step.
+- Routing should decide:
+  - 2-way match
+  - 3-way match
+  - contract match
+  - no match
+- Keep match routing separate from match logic.
+- Keep each match type in its own file.
+- The matching orchestrator should only connect routing and match modules.
+- Do not put matching business logic inside the orchestrator.
+- Final output should include validation result + matching result.

src/Invoice Validation Functions/Tax - ID validation.py

@@ -0,0 +1,564 @@
+import re
+from datetime import datetime, date, timezone
+from hashlib import sha256
+from unicodedata import normalize
+
+
+def _h_gb_vat_mod97(value, **_kw):
+    # UK VAT mod-97. Tries Method A then Method B (new-style) per HMRC spec.
+    # Value: 9 or 12 digits, GB prefix already stripped.
+    digits = re.sub(r"\D", "", value)
+    if len(digits) not in (9, 12):
+        return False
+    d = [int(c) for c in digits[:9]]
+    weights = [8, 7, 6, 5, 4, 3, 2]
+    s = sum(weights[i] * d[i] for i in range(7))
+    check = d[7] * 10 + d[8]
+    return (s - check) % 97 == 0 or (s + 55 - check) % 97 == 0
+
+
+def _h_de_vat_mod11(value, **_kw):
+    # German VAT recursive mod-11 check digit.
+    # Value: 9 digits, DE prefix already stripped.
+    digits = re.sub(r"\D", "", value)
+    if len(digits) != 9:
+        return False
+    product = 10
+    for c in digits[:8]:
+        s = (int(c) + product) % 10
+        if s == 0:
+            s = 10
+        product = (2 * s) % 11
+    check = 11 - product
+    if check == 10:
+        check = 0
+    return int(digits[8]) == check
+
+
+def _h_fr_vat_mod97(value, **_kw):
+    # French TVA key check.
+    # Value: 11 chars (FR prefix stripped): 2-char key + 9-digit SIREN.
+    # Numeric keys use the standard formula; alphanumeric keys use base-36 encoding.
+    if len(value) != 11:
+        return False
+    siren_str = value[2:]
+    if not siren_str.isdigit():
+        return False
+    siren = int(siren_str)
+    key = value[:2]
+    if key.isdigit():
+        return int(key) == (12 + 3 * (siren % 97)) % 97
+    def _char_val(c):
+        return int(c) if c.isdigit() else ord(c.upper()) - 55
+    k = _char_val(key[0]) * 36 + _char_val(key[1])
+    return k == (24 + 25 * (siren % 97)) % 97
+
+
+def _h_it_vat_luhn(value, **_kw):
+    # Italian Partita IVA Luhn-style check digit.
+    # Value: 11 digits, IT prefix already stripped.
+    digits = re.sub(r"\D", "", value)
+    if len(digits) != 11:
+        return False
+    odd_sum = sum(int(digits[i]) for i in range(0, 10, 2))
+    even_sum = 0
+    for i in range(1, 10, 2):
+        d = int(digits[i]) * 2
+        if d > 9:
+            d -= 9
+        even_sum += d
+    total = odd_sum + even_sum
+    return int(digits[10]) == (10 - total % 10) % 10
+
+
+def _h_es_nif(value, **_kw):
+    # Spanish NIF/CIF check character.
+    # Value: 9 chars (ES prefix stripped): [A-Z0-9][0-9]{7}[A-Z0-9].
+    if len(value) != 9:
+        return False
+    middle = value[1:8]
+    if not middle.isdigit():
+        return False
+    total = 0
+    for i, c in enumerate(middle):
+        d = int(c)
+        if i % 2 == 0:
+            d *= 2
+            if d > 9:
+                d -= 9
+        total += d
+    control = (10 - total % 10) % 10
+    check = value[8]
+    if check.isdigit():
+        return int(check) == control
+    return check.upper() == "JABCDEFGHI"[control]
+
+
+def _h_gstin_crc(value, **_kw):
+    # Indian GSTIN Luhn-mod36 check digit (official GSTN algorithm).
+    # Value: 15-char GSTIN (no prefix removed — GSTIN has no country prefix in its pattern).
+    if len(value) != 15:
+        return False
+    charset = "0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZ"
+    total = 0
+    factor = 2
+    for c in value[:14]:
+        if c not in charset:
+            return False
+        pos = charset.index(c)
+        product = factor * pos
+        total += product // 36 + product % 36
+        factor = 3 - factor
+    check_pos = (36 - total % 36) % 36
+    return value[14] == charset[check_pos]
+
+
+def _h_abn_mod89(value, **_kw):
+    # Australian Business Number mod-89 weighted check.
+    # Value: 11 digits (no prefix removed).
+    digits = re.sub(r"\D", "", value)
+    if len(digits) != 11:
+        return False
+    weights = [10, 1, 3, 5, 7, 9, 11, 13, 15, 17, 19]
+    d = [int(c) for c in digits]
+    d[0] -= 1
+    return sum(d[i] * weights[i] for i in range(11)) % 89 == 0
+
+
+def _h_bn_luhn(value, **_kw):
+    # Canadian Business Number Luhn check on the 9-digit registration number.
+    # The optional 6-char program account suffix (if present) is ignored.
+    nine = value[:9]
+    if len(nine) != 9 or not nine.isdigit():
+        return False
+    total = 0
+    for i in range(9):
+        d = int(nine[i])
+        if i % 2 == 1:
+            d *= 2
+            if d > 9:
+                d -= 9
+        total += d
+    return total % 10 == 0
+
+
+def _h_sg_nric_fin(value, **_kw):
+    # Singapore NRIC/FIN check character.
+    # Value: 9 chars — [STFG][0-9]{7}[A-Z] (no prefix removed).
+    value = value.upper()
+    if len(value) != 9 or value[0] not in "STFG" or not value[1:8].isdigit():
+        return False
+    weights = [2, 7, 6, 5, 4, 3, 2]
+    total = sum(int(value[i + 1]) * weights[i] for i in range(7))
+    if value[0] in "TG":
+        total += 4
+    check_chars = "JZIHGFEDCBA" if value[0] in "ST" else "XWUTRQPNMLK"
+    return value[8] == check_chars[total % 11]
+
+
+_BUILTIN_CHECKSUM_HANDLERS = {
+    "gb_vat_mod97": _h_gb_vat_mod97,
+    "de_vat_mod11": _h_de_vat_mod11,
+    "fr_vat_mod97": _h_fr_vat_mod97,
+    "it_vat_luhn": _h_it_vat_luhn,
+    "es_nif": _h_es_nif,
+    "gstin_crc": _h_gstin_crc,
+    "abn_mod89": _h_abn_mod89,
+    "bn_luhn": _h_bn_luhn,
+    "sg_nric_fin": _h_sg_nric_fin,
+}
+
+
+def run_tax_id_validation(
+    *,
+    document,
+    extracted_fields,
+    vendor,
+    vendor_tax_ids,
+    resolved_vendor_id,
+    platform_configs=None,
+):
+    """
+    Runs Tax ID Validation after vendor resolution.
+
+    Expected document fields:
+        document.id
+        document.tenant_id
+
+    Expected extracted_fields fields:
+        sender_tax_id
+        invoice_date
+
+    Expected vendor fields:
+        vendor.id
+        vendor.billing_country
+        vendor.tax_registration_number
+
+    Expected vendor_tax_ids row fields:
+        tenant_id
+        vendor_id
+        tax_id_canonical
+        tax_id_type
+        verified_at
+        deleted_at
+        effective_from
+        effective_to
+    """
+
+    platform_configs = platform_configs or {}
+
+    tax_id_registry = platform_configs.get(
+        "validation.tax_id_family_registry",
+        {
+            "VAT-GB": {"pattern": r"^(GB)?[0-9]{9}([0-9]{3})?$", "checksum": "gb_vat_mod97"},
+            "VAT-DE": {"pattern": r"^(DE)?[0-9]{9}$", "checksum": "de_vat_mod11"},
+            "VAT-FR": {"pattern": r"^(FR)?[A-Z0-9]{2}[0-9]{9}$", "checksum": "fr_vat_mod97"},
+            "VAT-IT": {"pattern": r"^(IT)?[0-9]{11}$", "checksum": "it_vat_luhn"},
+            "VAT-ES": {"pattern": r"^(ES)?[A-Z0-9][0-9]{7}[A-Z0-9]$", "checksum": "es_nif"},
+            "VAT-NL": {"pattern": r"^(NL)?[0-9]{9}B[0-9]{2}$", "checksum": None},
+            "GSTIN-IN": {
+                "pattern": r"^[0-9]{2}[A-Z]{5}[0-9]{4}[A-Z][1-9A-Z]Z[0-9A-Z]$",
+                "checksum": "gstin_crc",
+            },
+            "PAN-IN": {"pattern": r"^[A-Z]{5}[0-9]{4}[A-Z]$", "checksum": None},
+            "EIN-US": {"pattern": r"^[0-9]{2}-?[0-9]{7}$", "checksum": None},
+            "ABN-AU": {"pattern": r"^[0-9]{11}$", "checksum": "abn_mod89"},
+            "BN-CA": {"pattern": r"^[0-9]{9}([A-Z]{2}[0-9]{4})?$", "checksum": "bn_luhn"},
+            "TIN-SG": {"pattern": r"^(?:[STFG][0-9]{7}[A-Z]|[0-9]{9}[A-Z])$", "checksum": "sg_nric_fin"},
+        },
+    )
+
+    # Platform can override built-in handlers via validation.tax_id_checksum_handlers.
+    # Each handler: callable(value: str, tax_id_type: str | None, country_code: str | None) -> bool
+    checksum_handlers = platform_configs.get("validation.tax_id_checksum_handlers", _BUILTIN_CHECKSUM_HANDLERS)
+
+    def hash_prefix(value, length=8):
+        if not value:
+            return None
+        return sha256(str(value).encode("utf-8")).hexdigest()[:length]
+
+    def normalise_tax_id(value):
+        if value is None:
+            return None
+
+        value = normalize("NFKC", str(value))
+        value = re.sub(r"[\s\-\.,/\\:;_]+", "", value)
+        value = value.upper().strip()
+        return value or None
+
+    def country_from_tax_id_type(tax_id_type):
+        if not tax_id_type or "-" not in tax_id_type:
+            return None
+        return tax_id_type.split("-")[-1]
+
+    def registry_entries_for_country(country_code):
+        if not country_code:
+            return list(tax_id_registry.items())
+
+        country_code = str(country_code).upper()
+        return [
+            (tax_id_type, config)
+            for tax_id_type, config in tax_id_registry.items()
+            if country_from_tax_id_type(tax_id_type) == country_code
+        ]
+
+    def remove_country_prefix(value, tax_id_type):
+        country_code = country_from_tax_id_type(tax_id_type)
+        if country_code and value.startswith(country_code):
+            return value[len(country_code):]
+        return value
+
+    def checksum_valid(value, checksum_name, tax_id_type=None, country_code=None):
+        """
+        Real checksum gate:
+        - if checksum is not configured, treat as valid
+        - if a handler is provided by platform configs, use it
+        - otherwise fail closed (False) so invalid IDs do not silently pass
+        """
+        if checksum_name is None:
+            return True
+
+        handler = checksum_handlers.get(checksum_name)
+        if callable(handler):
+            return bool(handler(value, tax_id_type=tax_id_type, country_code=country_code))
+
+        return False
+
+    def canonicalise_tax_id(value, id_type=None, country_code=None):
+        stripped = normalise_tax_id(value)
+
+        if not stripped:
+            return {
+                "canonical": None,
+                "inferred_id_type": None,
+                "country_code": None,
+                "format_valid": None,
+                "checksum_valid": None,
+                "validation_passed": None,
+            }
+
+        resolved_id_type = None
+        registry_config = None
+
+        # Master path: direct lookup on (id_type, country_code).
+        if id_type:
+            resolved_id_type = id_type
+            registry_config = tax_id_registry.get(id_type)
+            # Cross-check: country embedded in id_type must match supplied country_code.
+            if registry_config and country_code:
+                embedded = country_from_tax_id_type(id_type)
+                if embedded and str(embedded).upper() != str(country_code).upper():
+                    registry_config = None
+
+        # Invoice path: infer by trying registry patterns for the supplied country.
+        else:
+            for candidate_id_type, candidate_config in registry_entries_for_country(country_code):
+                if re.match(candidate_config["pattern"], stripped):
+                    resolved_id_type = candidate_id_type
+                    registry_config = candidate_config
+                    break
+
+        # No registry entry matched: return stripped canonical value, no format/checksum enforcement.
+        if not registry_config:
+            return {
+                "canonical": stripped,
+                "inferred_id_type": None,
+                "country_code": None,
+                "format_valid": None,
+                "checksum_valid": None,
+                "validation_passed": None,
+            }
+
+        canonical = remove_country_prefix(stripped, resolved_id_type)
+        format_valid = bool(re.match(registry_config["pattern"], stripped))
+        checksum_ok = checksum_valid(
+            canonical,
+            registry_config.get("checksum"),
+            tax_id_type=resolved_id_type,
+            country_code=country_from_tax_id_type(resolved_id_type),
+        )
+
+        return {
+            "canonical": canonical,
+            "inferred_id_type": resolved_id_type,
+            "country_code": country_from_tax_id_type(resolved_id_type),
+            "format_valid": format_valid,
+            "checksum_valid": checksum_ok,
+            "validation_passed": bool(format_valid and checksum_ok),
+        }
+
+    def is_effective(row, invoice_date):
+        if getattr(row, "deleted_at", None) is not None:
+            return False
+
+        if invoice_date is None:
+            return False
+
+        effective_from = getattr(row, "effective_from", None)
+        effective_to = getattr(row, "effective_to", None)
+
+        if effective_from and invoice_date < effective_from:
+            return False
+
+        if effective_to and invoice_date >= effective_to:
+            return False
+
+        return True
+
+    risk_flags = []
+    advisory_flags = []
+    skipped_steps = []
+
+    if resolved_vendor_id is None:
+        skipped_steps.append(
+            {
+                "step": "tax_id_validation",
+                "skip_reason": "vendor_unresolved",
+            }
+        )
+        return {
+            "risk_flags": risk_flags,
+            "advisory_flags": advisory_flags,
+            "skipped_steps": skipped_steps,
+            "audit_payload": None,
+            "match_result": "skipped",
+        }
+
+    invoice_tax_id = getattr(extracted_fields, "sender_tax_id", None)
+    if not invoice_tax_id:
+        skipped_steps.append(
+            {
+                "step": "tax_id_validation",
+                "skip_reason": "no_invoice_tax_id",
+            }
+        )
+        return {
+            "risk_flags": risk_flags,
+            "advisory_flags": advisory_flags,
+            "skipped_steps": skipped_steps,
+            "audit_payload": None,
+            "match_result": "skipped",
+        }
+
+    invoice_date = getattr(extracted_fields, "invoice_date", None)
+    if isinstance(invoice_date, datetime):
+        invoice_date = invoice_date.date()
+    elif invoice_date is not None and not isinstance(invoice_date, date):
+        # Do not invent today's date; treat unknown types as unavailable.
+        invoice_date = None
+
+    vendor_country = getattr(vendor, "billing_country", None)
+
+    invoice_canonical_result = canonicalise_tax_id(
+        invoice_tax_id,
+        id_type=None,
+        country_code=vendor_country,
+    )
+
+    invoice_canonical = invoice_canonical_result["canonical"]
+    invoice_inferred_id_type = invoice_canonical_result["inferred_id_type"]
+    invoice_inferred_country = invoice_canonical_result["country_code"]
+
+    # §7.3.3: unconstrained inference to detect cross-country format mismatch.
+    # The constrained call above only tries patterns for vendor_country, so its
+    # inferred_country always equals vendor_country (or is None). A separate
+    # country_code=None call tries all registry entries and can identify that the
+    # invoice ID looks like a different country's format.
+    _global_infer = canonicalise_tax_id(invoice_tax_id, id_type=None, country_code=None)
+    _global_inferred_country = _global_infer["country_code"]
+
+    if not invoice_canonical:
+        skipped_steps.append({"step": "tax_id_validation", "skip_reason": "no_invoice_tax_id"})
+        return {
+            "risk_flags": risk_flags,
+            "advisory_flags": advisory_flags,
+            "skipped_steps": skipped_steps,
+            "audit_payload": None,
+            "match_result": "skipped",
+        }
+
+    candidate_rows_for_vendor = [
+        row
+        for row in vendor_tax_ids
+        if getattr(row, "tenant_id", None) == document.tenant_id
+        and getattr(row, "vendor_id", None) == resolved_vendor_id
+        and getattr(row, "deleted_at", None) is None
+    ]
+
+    effective_candidate_rows = [
+        row
+        for row in candidate_rows_for_vendor
+        if is_effective(row, invoice_date)
+    ]
+
+    matching_vendor_tax_row = None
+    for row in effective_candidate_rows:
+        if getattr(row, "tax_id_canonical", None) == invoice_canonical:
+            matching_vendor_tax_row = row
+            break
+
+    master_fallback_result = None
+    fallback_master_canonical = None
+
+    if not candidate_rows_for_vendor and getattr(vendor, "tax_registration_number", None):
+        master_fallback_result = canonicalise_tax_id(
+            vendor.tax_registration_number,
+            id_type=None,
+            country_code=vendor_country,
+        )
+        fallback_master_canonical = master_fallback_result["canonical"]
+
+    match_result = "unmatched"
+    master_hash_prefix = None
+    audit_id_type = invoice_inferred_id_type
+
+    if matching_vendor_tax_row:
+        match_result = "matched"
+        master_hash_prefix = hash_prefix(matching_vendor_tax_row.tax_id_canonical)
+        audit_id_type = getattr(matching_vendor_tax_row, "tax_id_type", None)
+
+        if getattr(matching_vendor_tax_row, "verified_at", None) is None:
+            advisory_flags.append(
+                {
+                    "code": "tax_id_unverified_master",
+                    "severity": "advisory",
+                    "vendor_id": resolved_vendor_id,
+                }
+            )
+
+        # Country mismatch only after a real successful match path.
+        if (
+            _global_inferred_country
+            and vendor_country
+            and str(_global_inferred_country).upper() != str(vendor_country).upper()
+        ):
+            risk_flags.append(
+                {
+                    "code": "tax_id_country_mismatch",
+                    "severity": "medium",
+                    "weight": 0.25,
+                    "invoice_country": _global_inferred_country,
+                    "vendor_country": vendor_country,
+                }
+            )
+
+    elif not candidate_rows_for_vendor and fallback_master_canonical:
+        # Fallback only when there are no structured vendor_tax_ids rows at all.
+        master_hash_prefix = hash_prefix(fallback_master_canonical)
+
+        if invoice_canonical == fallback_master_canonical:
+            match_result = "matched_fallback"
+            if (
+                _global_inferred_country
+                and vendor_country
+                and str(_global_inferred_country).upper() != str(vendor_country).upper()
+            ):
+                risk_flags.append(
+                    {
+                        "code": "tax_id_country_mismatch",
+                        "severity": "medium",
+                        "weight": 0.25,
+                        "invoice_country": _global_inferred_country,
+                        "vendor_country": vendor_country,
+                    }
+                )
+        else:
+            risk_flags.append(
+                {
+                    "code": "tax_id_mismatch",
+                    "severity": "high",
+                    "vendor_id": resolved_vendor_id,
+                }
+            )
+
+    else:
+        risk_flags.append(
+            {
+                "code": "tax_id_mismatch",
+                "severity": "high",
+                "vendor_id": resolved_vendor_id,
+            }
+        )
+
+    audit_payload = {
+        "document_id": document.id,
+        "vendor_id": resolved_vendor_id,
+        "match_result": match_result,
+        "invoice_hash_prefix": hash_prefix(invoice_canonical),
+        "master_hash_prefix": master_hash_prefix,
+        "id_type": audit_id_type,
+        "country_code": invoice_inferred_country or vendor_country,
+        "called_at": datetime.now(timezone.utc).isoformat(),
+    }
+
+    return {
+        "risk_flags": risk_flags,
+        "advisory_flags": advisory_flags,
+        "skipped_steps": skipped_steps,
+        "audit_payload": audit_payload,
+        "match_result": match_result,
+        "invoice_tax_id_canonical": invoice_canonical,
+        "invoice_tax_id_inferred_type": invoice_inferred_id_type,
+        "invoice_tax_id_format_valid": invoice_canonical_result.get("format_valid"),
+        "invoice_tax_id_checksum_valid": invoice_canonical_result.get("checksum_valid"),
+    }

src/Invoice Validation Functions/Vendor Validation.py

@@ -0,0 +1,387 @@
+from decimal import Decimal
+from hashlib import sha256
+from unicodedata import normalize
+from uuid import uuid4
+from datetime import datetime, timezone
+
+from rapidfuzz import fuzz, process
+
+
+def run_vendor_validation(
+    *,
+    document,
+    extracted_fields,
+    vendors,
+    platform_configs,
+    route_contexts=None,
+    actor_id=None,
+):
+    """
+    Runs Cat 2 vendor validation using RapidFuzz instead of the platform name-match algorithm.
+
+    Expected document fields:
+        document.id
+        document.tenant_id
+
+    Expected extracted_fields fields:
+        sender_name
+        sender_name_canonical
+        sender_tax_id_canonical
+        bank_iban_canonical
+
+    Expected vendor fields:
+        id
+        tenant_id
+        legal_name
+        vendor_name
+        legal_name_canonical
+        billing_country
+        billing_address
+        vendor_status
+        deleted_at
+
+    Returns a dict containing:
+        resolved_vendor_id
+        resolved_sender_name
+        resolved_sender_addr
+        risk_flags
+        advisory_flags
+        skipped_steps
+        audit_row
+        match_result
+    """
+
+    route_contexts = set(route_contexts or [])
+
+    require_high_tier_contexts = set(
+        platform_configs.get(
+            "validation.vendor_match_require_high_tier_contexts",
+            ["contract_match", "bank_validation", "approval_revalidation"],
+        )
+    )
+
+    algorithm_version = platform_configs.get(
+        "validation.name_match_algorithm_version",
+        "rapidfuzz.local.v1",
+    )
+
+    high_threshold = Decimal(str(platform_configs.get("validation.rapidfuzz_high_threshold", "0.90")))
+    medium_threshold = Decimal(str(platform_configs.get("validation.rapidfuzz_medium_threshold", "0.75")))
+    suggestion_threshold = Decimal(str(platform_configs.get("validation.rapidfuzz_suggestion_threshold", "0.60")))
+
+    def canonicalize(value):
+        if not value:
+            return ""
+        return normalize("NFKC", str(value)).lower().strip()
+
+    def hash_prefix(value, length=8):
+        if not value:
+            return None
+        return sha256(str(value).encode("utf-8")).hexdigest()[:length]
+
+    def get_vendor_names(vendor):
+        names = []
+
+        for name in (
+            getattr(vendor, "legal_name", None),
+            getattr(vendor, "vendor_name", None),
+        ):
+            canonical = canonicalize(name)
+            if canonical and canonical not in names:
+                names.append(canonical)
+
+        return names
+
+    def build_candidate_pool():
+        pool = []
+
+        for vendor in vendors:
+            if getattr(vendor, "tenant_id", None) != document.tenant_id:
+                continue
+
+            if getattr(vendor, "vendor_status", None) == "deleted":
+                continue
+
+            if getattr(vendor, "deleted_at", None) is not None:
+                continue
+
+            names = get_vendor_names(vendor)
+            if not names:
+                continue
+
+            pool.append(
+                {
+                    "record_id": vendor.id,
+                    "vendor": vendor,
+                    "names": names,
+                    "auxiliary": {
+                        "country": getattr(vendor, "billing_country", None),
+                        "tax_id_hash": hash_prefix(
+                            getattr(extracted_fields, "sender_tax_id_canonical", None)
+                        ),
+                        "iban_hash": hash_prefix(
+                            getattr(extracted_fields, "bank_iban_canonical", None)
+                        ),
+                    },
+                }
+            )
+
+        return pool
+
+    def rapidfuzz_name_match(query_name, candidate_pool):
+        choices = []
+
+        for candidate in candidate_pool:
+            for name in candidate["names"]:
+                choices.append(
+                    {
+                        "name": name,
+                        "record_id": candidate["record_id"],
+                        "vendor": candidate["vendor"],
+                    }
+                )
+
+        if not query_name or not choices:
+            return {
+                "decision": "unmatched",
+                "matched_record_id": None,
+                "matched_vendor": None,
+                "runner_up_record_id": None,
+                "runner_up_vendor": None,
+                "score": Decimal("0"),
+                "score_tier": "low",
+                "algorithm_version": algorithm_version,
+                "trace_id": str(uuid4()),
+            }
+
+        raw_matches = process.extract(
+            query_name,
+            [choice["name"] for choice in choices],
+            scorer=fuzz.WRatio,
+            limit=5,
+        )
+
+        ranked = []
+        seen_record_ids = set()
+
+        for matched_name, score, choice_index in raw_matches:
+            choice = choices[choice_index]
+            record_id = choice["record_id"]
+
+            if record_id in seen_record_ids:
+                continue
+
+            seen_record_ids.add(record_id)
+
+            ranked.append(
+                {
+                    "record_id": record_id,
+                    "vendor": choice["vendor"],
+                    "matched_name": matched_name,
+                    "score": Decimal(str(score / 100)).quantize(Decimal("0.0001")),
+                }
+            )
+
+        if not ranked:
+            best = None
+            runner_up = None
+            score = Decimal("0")
+        else:
+            best = ranked[0]
+            runner_up = ranked[1] if len(ranked) > 1 else None
+            score = best["score"]
+
+        if score >= high_threshold:
+            decision = "matched"
+            score_tier = "high"
+            matched_record_id = best["record_id"]
+            matched_vendor = best["vendor"]
+        elif score >= medium_threshold:
+            decision = "matched"
+            score_tier = "medium"
+            matched_record_id = best["record_id"]
+            matched_vendor = best["vendor"]
+        elif score >= suggestion_threshold:
+            decision = "suggestion"
+            score_tier = "low"
+            matched_record_id = None
+            matched_vendor = None
+        else:
+            decision = "unmatched"
+            score_tier = "low"
+            matched_record_id = None
+            matched_vendor = None
+
+        return {
+            "decision": decision,
+            "matched_record_id": matched_record_id,
+            "matched_vendor": matched_vendor,
+            "runner_up_record_id": (
+                best["record_id"] if decision == "suggestion" and best
+                else (runner_up["record_id"] if runner_up else None)
+            ),
+            "runner_up_vendor": (
+                best["vendor"] if decision == "suggestion" and best
+                else (runner_up["vendor"] if runner_up else None)
+            ),
+            "score": score,
+            "score_tier": score_tier,
+            "algorithm_version": algorithm_version,
+            "trace_id": str(uuid4()),
+        }
+
+    risk_flags = []
+    advisory_flags = []
+    skipped_steps = []
+
+    resolved_vendor_id = None
+    resolved_sender_name = None
+    resolved_sender_addr = None
+
+    query_name = canonicalize(
+        getattr(extracted_fields, "sender_name_canonical", None)
+        or getattr(extracted_fields, "sender_name", None)
+    )
+
+    candidate_pool = build_candidate_pool()
+
+    match_result = rapidfuzz_name_match(
+        query_name=query_name,
+        candidate_pool=candidate_pool,
+    )
+
+    decision = match_result["decision"]
+    score_tier = match_result["score_tier"]
+
+    control_context_requires_high = bool(route_contexts & require_high_tier_contexts)
+
+    if decision == "matched" and score_tier == "high":
+        resolved_vendor_id = match_result["matched_record_id"]
+
+    elif decision == "matched" and score_tier == "medium":
+        if control_context_requires_high:
+            match_result["decision"] = "suggestion"
+            advisory_flags.append(
+                {
+                    "code": "vendor_suggestion",
+                    "severity": "advisory",
+                    "record_id": match_result["matched_record_id"],
+                    "reason": "medium_tier_match_downgraded_for_control_context",
+                }
+            )
+            skipped_steps.extend(
+                [
+                    {"step": "vendor_tax_validation", "skip_reason": "vendor_unresolved"},
+                    {"step": "vendor_bank_validation", "skip_reason": "vendor_unresolved"},
+                    {"step": "vendor_status_validation", "skip_reason": "vendor_unresolved"},
+                ]
+            )
+        else:
+            resolved_vendor_id = match_result["matched_record_id"]
+            risk_flags.append(
+                {
+                    "code": "low_confidence_vendor_resolution",
+                    "severity": "medium",
+                    "record_id": resolved_vendor_id,
+                    "score_tier": score_tier,
+                }
+            )
+
+    elif decision == "suggestion":
+        advisory_flags.append(
+            {
+                "code": "vendor_suggestion",
+                "severity": "advisory",
+                "record_id": match_result["runner_up_record_id"],
+                "reason": "possible_vendor_match",
+            }
+        )
+        skipped_steps.extend(
+            [
+                {"step": "sender_field_override", "skip_reason": "vendor_unresolved"},
+                {"step": "vendor_tax_validation", "skip_reason": "vendor_unresolved"},
+                {"step": "vendor_bank_validation", "skip_reason": "vendor_unresolved"},
+                {"step": "vendor_status_validation", "skip_reason": "vendor_unresolved"},
+            ]
+        )
+
+    else:
+        risk_flags.append(
+            {
+                "code": "unknown_vendor",
+                "severity": "high",
+                "reason": "no_vendor_match_found",
+            }
+        )
+        skipped_steps.extend(
+            [
+                {"step": "sender_field_override", "skip_reason": "vendor_unresolved"},
+                {"step": "vendor_tax_validation", "skip_reason": "vendor_unresolved"},
+                {"step": "vendor_bank_validation", "skip_reason": "vendor_unresolved"},
+                {"step": "vendor_status_validation", "skip_reason": "vendor_unresolved"},
+            ]
+        )
+
+    blocked_vendor = None
+
+    if match_result["decision"] in ("matched", "suggestion"):
+        if match_result.get("matched_vendor") and match_result["matched_vendor"].vendor_status == "blocked":
+            blocked_vendor = match_result["matched_vendor"]
+
+        if match_result.get("runner_up_vendor") and match_result["runner_up_vendor"].vendor_status == "blocked":
+            blocked_vendor = match_result["runner_up_vendor"]
+
+    if blocked_vendor:
+        risk_flags.append(
+            {
+                "code": "vendor_blocked",
+                "severity": "high",
+                "record_id": blocked_vendor.id,
+                "reason": "matched_or_suggested_vendor_is_blocked",
+            }
+        )
+
+    if resolved_vendor_id:
+        resolved_vendor = match_result["matched_vendor"]
+
+        resolved_sender_name = (
+            getattr(resolved_vendor, "legal_name", None)
+            or getattr(resolved_vendor, "vendor_name", None)
+        )
+
+        billing_address = getattr(resolved_vendor, "billing_address", None)
+        if billing_address is not None:
+            resolved_sender_addr = billing_address
+
+    audit_row = {
+        "document_id": document.id,
+        "query_name_hash_prefix_8": hash_prefix(query_name, length=8),
+        "algorithm_version": algorithm_version,
+        "decision": match_result["decision"],
+        "matched_record_id": match_result["matched_record_id"],
+        "runner_up_record_id": match_result["runner_up_record_id"],
+        "score_tier": match_result["score_tier"],
+        "score": str(match_result["score"]),
+        "trace_id": match_result["trace_id"],
+        "called_at": datetime.now(timezone.utc),
+        "actor_id": actor_id,
+    }
+
+    return {
+        "resolved_vendor_id": resolved_vendor_id,
+        "resolved_sender_name": resolved_sender_name,
+        "resolved_sender_addr": resolved_sender_addr,
+        "risk_flags": risk_flags,
+        "advisory_flags": advisory_flags,
+        "skipped_steps": skipped_steps,
+        "audit_row": audit_row,
+        "match_result": {
+            "decision": match_result["decision"],
+            "matched_record_id": resolved_vendor_id,
+            "runner_up_record_id": match_result["runner_up_record_id"],
+            "score": match_result["score"],
+            "score_tier": match_result["score_tier"],
+            "algorithm_version": match_result["algorithm_version"],
+            "trace_id": match_result["trace_id"],
+        },
+    }

src/Invoice Validation Functions/completeness.py

@@ -0,0 +1,171 @@
+
+from datetime import date, datetime
+from decimal import Decimal, InvalidOperation
+
+
+def parse_date(value):
+    if value in (None, ""):
+        return None
+    if isinstance(value, datetime):
+        return value.date()
+    if isinstance(value, date):
+        return value
+    try:
+        return datetime.strptime(str(value), "%Y-%m-%d").date()
+    except ValueError:
+        return None
+
+
+def parse_decimal(value):
+    if value in (None, ""):
+        return None
+    if isinstance(value, Decimal):
+        return value
+    try:
+        return Decimal(str(value))
+    except (InvalidOperation, ValueError):
+        return None
+
+
+def run_invoice_validation(*, document, extracted_fields, line_items, currency_seed_map, platform_configs=None):
+    """
+    Business validation logic for:
+      7.7.1 Required fields + due-date derivation path
+      7.7.2 Currency membership
+      7.7.3 Math cross-check with currency-aware tolerance
+    """
+    platform_configs = platform_configs or {}
+
+    cap_minor_units = int(
+        platform_configs.get("validation.solver_tolerance.cap_minor_units", 0)
+    )
+    base_per_minor_unit = int(
+        platform_configs.get("validation.solver_tolerance.base_per_minor_unit", 0)
+    )
+    per_line_minor_units = int(
+        platform_configs.get("validation.solver_tolerance.per_line_minor_units", 0)
+    )
+
+    risk_flags = []
+    advisory_flags = []
+    skipped_steps = []
+
+    def add_flag(code, *, weight, detail=None):
+        risk_flags.append(
+            {
+                "code": code,
+                "severity": "advisory",
+                "weight": weight,
+                "detail": detail or {},
+            }
+        )
+
+    invoice_no = getattr(extracted_fields, "invoice_no", None)
+    invoice_date = parse_date(getattr(extracted_fields, "invoice_date", None))
+    total_amount = parse_decimal(getattr(extracted_fields, "total_amount", None))
+    subtotal = parse_decimal(getattr(extracted_fields, "subtotal", None))
+    sender_name = getattr(extracted_fields, "sender_name", None)
+    currency = getattr(extracted_fields, "currency", None)
+    due_date = parse_date(getattr(extracted_fields, "due_date", None))
+    payment_terms = getattr(extracted_fields, "payment_terms", None)
+
+    missing_fields = []
+
+    def mark_missing(field_name):
+        if field_name not in missing_fields:
+            missing_fields.append(field_name)
+
+    if not invoice_no:
+        mark_missing("invoice_no")
+    if invoice_date is None:
+        mark_missing("invoice_date")
+    if total_amount is None:
+        mark_missing("total_amount")
+    if not sender_name:
+        mark_missing("sender_name")
+    if not currency:
+        mark_missing("currency")
+
+    due_date_derivable = invoice_date is not None and payment_terms not in (None, "")
+    if due_date is None and not due_date_derivable:
+        mark_missing("due_date")
+        if invoice_date is None:
+            mark_missing("invoice_date")
+        if payment_terms in (None, ""):
+            mark_missing("payment_terms")
+
+    if missing_fields:
+        add_flag(
+            "incomplete_fields",
+            weight=0.25,
+            detail={"missing_fields": missing_fields},
+        )
+
+    currency_key = str(currency).strip().upper() if currency not in (None, "") else None
+    currency_row = currency_seed_map.get(currency_key) if currency_key else None
+
+    if currency_row is None:
+        add_flag(
+            "invalid_currency",
+            weight=0.20,
+            detail={
+                "currency": currency,
+                "reason": "currency_missing_or_not_in_iso_currencies_seed",
+            },
+        )
+
+    line_amounts = []
+    for li in line_items:
+        amount = parse_decimal(getattr(li, "amount", None))
+        if amount is not None:
+            line_amounts.append(amount)
+
+    num_line_items = len(line_amounts)
+    math_ran = False
+
+    if subtotal is not None and num_line_items > 0:
+        if currency_row is None:
+            skipped_steps.append(
+                {"step": "math_cross_check", "skip_reason": "currency_minor_unit_unknown"}
+            )
+        else:
+            minor_unit_value = parse_decimal(currency_row.get("minor_unit_value"))
+            if minor_unit_value is None:
+                skipped_steps.append(
+                    {"step": "math_cross_check", "skip_reason": "minor_unit_value_missing_in_seed"}
+                )
+            else:
+                math_ran = True
+                line_sum = sum(line_amounts, Decimal("0"))
+                tolerance_minor_units = min(
+                    cap_minor_units,
+                    base_per_minor_unit + per_line_minor_units * num_line_items,
+                )
+                tolerance_decimal = Decimal(tolerance_minor_units) * minor_unit_value
+                difference = abs(line_sum - subtotal)
+
+                if difference > tolerance_decimal:
+                    add_flag(
+                        "math_error_line_items",
+                        weight=0.20,
+                        detail={
+                            "subtotal": str(subtotal),
+                            "line_items_sum": str(line_sum),
+                            "difference": str(difference),
+                            "tolerance_minor_units": tolerance_minor_units,
+                            "tolerance_decimal": str(tolerance_decimal),
+                            "currency": currency_key,
+                            "minor_unit_value": str(minor_unit_value),
+                            "num_line_items": num_line_items,
+                        },
+                    )
+
+    return {
+        "risk_flags": risk_flags,
+        "advisory_flags": advisory_flags,
+        "skipped_steps": skipped_steps,
+        "result": "flagged" if risk_flags else "clean",
+        "missing_fields": missing_fields,
+        "currency_checked": currency_row is not None,
+        "math_checked": math_ran,
+    }

src/Invoice Validation Functions/date_sanity.py

@@ -0,0 +1,221 @@
+from datetime import date, datetime, timedelta
+from zoneinfo import ZoneInfo
+
+
+def run_date_sanity_checks(
+    *,
+    document,
+    extracted_fields,
+    platform_configs=None,
+):
+    """
+    Runs Date Sanity Checks after date normalisation.
+
+    Expected document fields:
+        document.id
+        document.tenant_id
+        Optional: document.tenant_timezone
+
+    Expected extracted_fields fields:
+        invoice_date
+        due_date
+        due_date_origin
+        payment_terms
+        service_start_date
+        service_end_date
+
+    Returns:
+        dict with advisory risk flags and audit-friendly result.
+    """
+
+    platform_configs = platform_configs or {}
+
+    default_timezone = platform_configs.get(
+        "validation.default_timezone",
+        getattr(document, "tenant_timezone", "UTC") or "UTC",
+    )
+
+    max_invoice_age_days = int(
+        platform_configs.get("validation.max_invoice_age_days", 365)
+    )
+
+    max_invoice_age_days = max(30, min(max_invoice_age_days, 1825))
+
+    service_end_grace_days = int(
+        platform_configs.get("validation.service_end_post_invoice_grace_days", 0)
+    )
+
+    def as_date(value):
+        if value is None or value == "":
+            return None
+
+        if isinstance(value, datetime):
+            return value.date()
+
+        if isinstance(value, date):
+            return value
+
+        return datetime.strptime(str(value), "%Y-%m-%d").date()
+
+    def as_int(value):
+        if value is None or value == "":
+            return None
+        try:
+            return int(value)
+        except (ValueError, TypeError):
+            return None
+
+    def tenant_current_date():
+        try:
+            return datetime.now(ZoneInfo(default_timezone)).date()
+        except Exception:
+            return datetime.now(ZoneInfo("UTC")).date()
+
+    def add_flag(code, *, weight, field, detail=None):
+        risk_flags.append(
+            {
+                "code": code,
+                "severity": "advisory",
+                "weight": weight,
+                "field": field,
+                "detail": detail or {},
+            }
+        )
+
+    risk_flags = []
+    advisory_flags = []
+    skipped_steps = []
+
+    invoice_date = as_date(getattr(extracted_fields, "invoice_date", None))
+    due_date = as_date(getattr(extracted_fields, "due_date", None))
+    service_start_date = as_date(getattr(extracted_fields, "service_start_date", None))
+    service_end_date = as_date(getattr(extracted_fields, "service_end_date", None))
+    due_date_origin = getattr(extracted_fields, "due_date_origin", None)
+    payment_terms = as_int(getattr(extracted_fields, "payment_terms", None))
+
+    current_date = tenant_current_date()
+
+    # 1. Payment terms must be non-negative.
+    # Zero is allowed: due-on-receipt.
+    if payment_terms is not None and payment_terms < 0:
+        add_flag(
+            "date_anomaly_payment_terms_negative",
+            weight=0.20,
+            field="payment_terms",
+            detail={
+                "payment_terms": payment_terms,
+            },
+        )
+
+    # 2. Due date must be on or after invoice date.
+    if invoice_date is not None and due_date is not None:
+        if due_date < invoice_date:
+            add_flag(
+                "date_anomaly_due_before_invoice",
+                weight=0.20,
+                field="due_date",
+                detail={
+                    "invoice_date": str(invoice_date),
+                    "due_date": str(due_date),
+                },
+            )
+
+    if invoice_date is None:
+        skipped_steps.extend([
+            {"step": "date_anomaly_invoice_future_dated", "skip_reason": "no_invoice_date"},
+            {"step": "date_anomaly_invoice_too_old", "skip_reason": "no_invoice_date"},
+        ])
+
+    # 3. Invoice must not be future-dated.
+    if invoice_date is not None and invoice_date > current_date:
+        add_flag(
+            "date_anomaly_invoice_future_dated",
+            weight=0.20,
+            field="invoice_date",
+            detail={
+                "invoice_date": str(invoice_date),
+                "current_date": str(current_date),
+                "timezone": default_timezone,
+            },
+        )
+
+    # 4. Invoice must not be too old.
+    if invoice_date is not None:
+        oldest_allowed_date = current_date - timedelta(days=max_invoice_age_days)
+
+        if invoice_date < oldest_allowed_date:
+            add_flag(
+                "date_anomaly_invoice_too_old",
+                weight=0.15,
+                field="invoice_date",
+                detail={
+                    "invoice_date": str(invoice_date),
+                    "oldest_allowed_date": str(oldest_allowed_date),
+                    "max_invoice_age_days": max_invoice_age_days,
+                },
+            )
+
+    # 5. Service period must not be reversed.
+    if service_start_date is not None and service_end_date is not None:
+        if service_end_date < service_start_date:
+            add_flag(
+                "date_anomaly_service_period_reversed",
+                weight=0.15,
+                field="service_end_date",
+                detail={
+                    "service_start_date": str(service_start_date),
+                    "service_end_date": str(service_end_date),
+                },
+            )
+
+    # 6. Service end must not be too far after invoice date.
+    if service_end_date is not None:
+        if invoice_date is None:
+            skipped_steps.append(
+                {"step": "date_anomaly_service_end_after_invoice", "skip_reason": "no_invoice_date"}
+            )
+        else:
+            allowed_service_end = invoice_date + timedelta(days=service_end_grace_days)
+
+            if service_end_date > allowed_service_end:
+                add_flag(
+                    "date_anomaly_service_end_after_invoice",
+                    weight=0.15,
+                    field="service_end_date",
+                    detail={
+                        "invoice_date": str(invoice_date),
+                        "service_end_date": str(service_end_date),
+                        "allowed_service_end": str(allowed_service_end),
+                        "grace_days": service_end_grace_days,
+                    },
+                )
+
+    # 7. Due date could not be derived.
+    if due_date is None and due_date_origin == "unknown":
+        add_flag(
+            "due_date_not_derivable",
+            weight=0.10,
+            field="due_date",
+            detail={
+                "due_date_origin": due_date_origin,
+            },
+        )
+
+    return {
+        "risk_flags": risk_flags,
+        "advisory_flags": advisory_flags,
+        "skipped_steps": skipped_steps,
+        "flag_codes": [flag["code"] for flag in risk_flags],
+        "checked_fields": {
+            "invoice_date": str(invoice_date) if invoice_date else None,
+            "due_date": str(due_date) if due_date else None,
+            "due_date_origin": due_date_origin,
+            "payment_terms": payment_terms,
+            "service_start_date": str(service_start_date) if service_start_date else None,
+            "service_end_date": str(service_end_date) if service_end_date else None,
+            "current_date": str(current_date),
+            "timezone": default_timezone,
+            "max_invoice_age_days": max_invoice_age_days,
+            "service_end_post_invoice_grace_days": service_end_grace_days,
+        },
+    }

src/Invoice Validation Functions/orchestrator.py

@@ -0,0 +1,282 @@
+"""
+Invoice validation orchestrator.
+
+Loads platform_configs.json once, runs all seven validators in SDD order,
+wires outputs to downstream inputs, and returns a single consolidated result.
+
+No business logic lives here — all decisions remain inside the validator modules.
+"""
+
+import importlib.util
+import json
+import os
+
+
+# ---------------------------------------------------------------------------
+# Module loader — handles filenames that contain spaces or hyphens
+# ---------------------------------------------------------------------------
+
+def _load_fn(filename, fn_name):
+    """Import a function from a .py file by path."""
+    here = os.path.dirname(os.path.abspath(__file__))
+    path = os.path.join(here, filename)
+    spec = importlib.util.spec_from_file_location(filename, path)
+    mod = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(mod)
+    return getattr(mod, fn_name)
+
+
+_run_vendor_validation        = _load_fn("Vendor Validation.py",         "run_vendor_validation")
+_run_tax_id_validation        = _load_fn("Tax - ID validation.py",        "run_tax_id_validation")
+_run_bank_validation          = _load_fn("Bank validation.py",            "run_bank_validation")
+_run_date_sanity_checks       = _load_fn("date_sanity.py",                "run_date_sanity_checks")
+_run_payment_terms_validation = _load_fn("payment terms valdiation.py",   "run_payment_terms_validation")
+_run_tax_amount_validation    = _load_fn("tax validation.py",             "run_tax_amount_validation")
+_run_completeness_validation  = _load_fn("completeness.py",               "run_invoice_validation")
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+class _FieldProxy:
+    """
+    Overlays computed attributes onto an existing extracted_fields object so
+    downstream validators can read Tax-ID results via the normal
+    getattr(extracted_fields, ...) pattern without mutating the original object.
+
+    Only non-None values are overlaid; None values fall through to the base
+    object (and its getattr default), which preserves the lenient defaults each
+    validator defines for fields it cannot always receive.
+    """
+    __slots__ = ("_base", "_overlay")
+
+    def __init__(self, base, **overlay):
+        object.__setattr__(self, "_base", base)
+        object.__setattr__(self, "_overlay", {k: v for k, v in overlay.items() if v is not None})
+
+    def __getattr__(self, name):
+        overlay = object.__getattribute__(self, "_overlay")
+        if name in overlay:
+            return overlay[name]
+        return getattr(object.__getattribute__(self, "_base"), name)
+
+
+def _load_platform_configs(path=None):
+    if path is None:
+        here = os.path.dirname(os.path.abspath(__file__))
+        path = os.path.join(here, "platform_configs.json")
+    with open(path, encoding="utf-8") as fh:
+        return json.load(fh)
+
+
+def _collect(result, key):
+    """Safely pull a list from a module result dict (returns [] if absent)."""
+    return result.get(key) or []
+
+
+# ---------------------------------------------------------------------------
+# Pipeline
+# ---------------------------------------------------------------------------
+
+def run_invoice_validation_pipeline(
+    *,
+    document,
+    extracted_fields,
+    vendors,
+    vendor_bank_accounts,
+    vendor_tax_ids,
+    entity,
+    line_items,
+    purchase_orders,
+    contracts,
+    tax_master_rows,
+    currency_seed_map,
+    platform_configs=None,
+    route_contexts=None,
+    actor_id=None,
+):
+    """
+    Run all seven validation modules in SDD order and return a consolidated result.
+
+    Parameters
+    ----------
+    document              : Invoice document object (needs .id, .tenant_id, .entity_id).
+    extracted_fields      : VLM-extracted fields object.
+    vendors               : Iterable of vendor records used for vendor matching.
+    vendor_bank_accounts  : Iterable of vendor bank account records.
+    vendor_tax_ids        : Iterable of vendor tax-ID records.
+    entity                : The routing entity object (needs .country_code, .vat_id, .region_code).
+    line_items            : Iterable of line-item records.
+    purchase_orders       : Iterable of purchase-order records.
+    contracts             : Iterable of contract records.
+    tax_master_rows       : Iterable of tax-master records.
+    currency_seed_map     : Dict mapping ISO currency code → row with minor_unit_value.
+    platform_configs      : Pre-loaded config dict; loaded from platform_configs.json when None.
+    route_contexts        : List of route-context strings forwarded to Vendor Validation.
+    actor_id              : Actor identifier forwarded to Vendor Validation for audit.
+
+    Returns
+    -------
+    {
+        "risk_flags":         [...],  # aggregated, in execution order
+        "advisory_flags":     [...],  # aggregated, in execution order
+        "skipped_steps":      [...],  # aggregated, in execution order
+        "ops_alerts":         [...],  # operational alerts (e.g. tax_master_unconfigured)
+        "resolved_vendor_id": ...,
+        "module_results":     {...},  # raw return dict from each module, keyed by name
+    }
+    """
+
+    if platform_configs is None:
+        platform_configs = _load_platform_configs()
+
+    risk_flags     = []
+    advisory_flags = []
+    skipped_steps  = []
+    ops_alerts     = []
+    module_results = {}
+
+    # ------------------------------------------------------------------
+    # Step 1 — Vendor Validation
+    # Runs first; produces resolved_vendor_id used by all downstream steps.
+    # ------------------------------------------------------------------
+    vendor_result = _run_vendor_validation(
+        document=document,
+        extracted_fields=extracted_fields,
+        vendors=vendors,
+        platform_configs=platform_configs,
+        route_contexts=route_contexts,
+        actor_id=actor_id,
+    )
+    module_results["vendor_validation"] = vendor_result
+    risk_flags.extend(_collect(vendor_result, "risk_flags"))
+    advisory_flags.extend(_collect(vendor_result, "advisory_flags"))
+    skipped_steps.extend(_collect(vendor_result, "skipped_steps"))
+
+    resolved_vendor_id = vendor_result.get("resolved_vendor_id")
+
+    # Resolve the vendor object so downstream modules receive it directly.
+    resolved_vendor = next(
+        (v for v in vendors if getattr(v, "id", None) == resolved_vendor_id),
+        None,
+    ) if resolved_vendor_id else None
+
+    # ------------------------------------------------------------------
+    # Step 2 — Tax ID Validation
+    # Depends on resolved_vendor_id and the resolved vendor object.
+    # ------------------------------------------------------------------
+    tax_id_result = _run_tax_id_validation(
+        document=document,
+        extracted_fields=extracted_fields,
+        vendor=resolved_vendor,
+        vendor_tax_ids=vendor_tax_ids,
+        resolved_vendor_id=resolved_vendor_id,
+        platform_configs=platform_configs,
+    )
+    module_results["tax_id_validation"] = tax_id_result
+    risk_flags.extend(_collect(tax_id_result, "risk_flags"))
+    advisory_flags.extend(_collect(tax_id_result, "advisory_flags"))
+    skipped_steps.extend(_collect(tax_id_result, "skipped_steps"))
+
+    # Enrich extracted_fields with Tax-ID results so Tax Validation can read
+    # sender_tax_id_inferred_family / _format_valid / _checksum_valid via its
+    # normal getattr(extracted_fields, ...) pattern.
+    enriched_fields = _FieldProxy(
+        extracted_fields,
+        sender_tax_id_inferred_family=tax_id_result.get("invoice_tax_id_inferred_type"),
+        sender_tax_id_format_valid=tax_id_result.get("invoice_tax_id_format_valid"),
+        sender_tax_id_checksum_valid=tax_id_result.get("invoice_tax_id_checksum_valid"),
+    )
+
+    # ------------------------------------------------------------------
+    # Step 3 — Bank Validation
+    # Depends on resolved_vendor_id.
+    # ------------------------------------------------------------------
+    bank_result = _run_bank_validation(
+        document=document,
+        extracted_fields=extracted_fields,
+        vendor_bank_accounts=vendor_bank_accounts,
+        resolved_vendor_id=resolved_vendor_id,
+        platform_configs=platform_configs,
+    )
+    module_results["bank_validation"] = bank_result
+    risk_flags.extend(_collect(bank_result, "risk_flags"))
+    advisory_flags.extend(_collect(bank_result, "advisory_flags"))
+    skipped_steps.extend(_collect(bank_result, "skipped_steps"))
+
+    # ------------------------------------------------------------------
+    # Step 4 — Date Sanity Checks
+    # Independent of vendor resolution.
+    # ------------------------------------------------------------------
+    date_result = _run_date_sanity_checks(
+        document=document,
+        extracted_fields=extracted_fields,
+        platform_configs=platform_configs,
+    )
+    module_results["date_sanity"] = date_result
+    risk_flags.extend(_collect(date_result, "risk_flags"))
+    advisory_flags.extend(_collect(date_result, "advisory_flags"))
+    skipped_steps.extend(_collect(date_result, "skipped_steps"))
+
+    # ------------------------------------------------------------------
+    # Step 5 — Payment Terms Validation
+    # Depends on resolved_vendor_id and the resolved vendor object.
+    # ------------------------------------------------------------------
+    payment_terms_result = _run_payment_terms_validation(
+        document=document,
+        extracted_fields=extracted_fields,
+        vendor=resolved_vendor,
+        purchase_orders=purchase_orders,
+        contracts=contracts,
+        resolved_vendor_id=resolved_vendor_id,
+        platform_configs=platform_configs,
+    )
+    module_results["payment_terms_validation"] = payment_terms_result
+    risk_flags.extend(_collect(payment_terms_result, "risk_flags"))
+    advisory_flags.extend(_collect(payment_terms_result, "advisory_flags"))
+    skipped_steps.extend(_collect(payment_terms_result, "skipped_steps"))
+
+    # ------------------------------------------------------------------
+    # Step 6 — Tax Amount Validation
+    # Depends on resolved vendor object and enriched_fields from Step 2.
+    # ------------------------------------------------------------------
+    tax_result = _run_tax_amount_validation(
+        document=document,
+        extracted_fields=enriched_fields,
+        line_items=line_items,
+        vendor=resolved_vendor,
+        entity=entity,
+        tax_master_rows=tax_master_rows,
+        platform_configs=platform_configs,
+    )
+    module_results["tax_validation"] = tax_result
+    risk_flags.extend(_collect(tax_result, "risk_flags"))
+    advisory_flags.extend(_collect(tax_result, "advisory_flags"))
+    skipped_steps.extend(_collect(tax_result, "skipped_steps"))
+    ops_alerts.extend(_collect(tax_result, "ops_alerts"))
+
+    # ------------------------------------------------------------------
+    # Step 7 — Completeness / Math Cross-check
+    # Independent of vendor resolution.
+    # ------------------------------------------------------------------
+    completeness_result = _run_completeness_validation(
+        document=document,
+        extracted_fields=extracted_fields,
+        line_items=line_items,
+        currency_seed_map=currency_seed_map,
+        platform_configs=platform_configs,
+    )
+    module_results["completeness"] = completeness_result
+    risk_flags.extend(_collect(completeness_result, "risk_flags"))
+    advisory_flags.extend(_collect(completeness_result, "advisory_flags"))
+    skipped_steps.extend(_collect(completeness_result, "skipped_steps"))
+
+    return {
+        "risk_flags": risk_flags,
+        "advisory_flags": advisory_flags,
+        "skipped_steps": skipped_steps,
+        "ops_alerts": ops_alerts,
+        "resolved_vendor_id": resolved_vendor_id,
+        "module_results": module_results,
+    }

src/Invoice Validation Functions/payment terms valdiation.py

@@ -0,0 +1,253 @@
+from datetime import date, datetime
+
+
+def run_payment_terms_validation(
+    *,
+    document,
+    extracted_fields,
+    vendor,
+    purchase_orders,
+    contracts,
+    resolved_vendor_id,
+    platform_configs=None,
+):
+    """
+    Payment Terms Validation.
+
+    Expected document fields:
+        document.id
+        document.tenant_id
+        document.entity_id
+
+    Expected extracted_fields fields:
+        payment_terms
+        po_number_canonical
+        invoice_date
+
+    Expected vendor fields:
+        vendor.id
+        vendor.payment_terms
+
+    Expected purchase_orders fields:
+        tenant_id
+        po_number_canonical
+        payment_terms
+        deleted_at
+
+    Expected contract fields:
+        tenant_id
+        vendor_id
+        entity_id
+        status
+        payment_terms
+        effective_from
+        effective_to
+        deleted_at
+
+    Returns:
+        dict with advisory flags, skipped status, and resolution detail.
+    """
+
+    platform_configs = platform_configs or {}
+
+    tolerance_days = int(
+        platform_configs.get("validation.payment_terms_tolerance_days", 0)
+    )
+
+    def as_int(value):
+        if value is None or value == "":
+            return None
+        try:
+            return int(value)
+        except (ValueError, TypeError):
+            return None
+
+    def as_date(value):
+        if value is None or value == "":
+            return None
+
+        if isinstance(value, datetime):
+            return value.date()
+
+        if isinstance(value, date):
+            return value
+
+        return datetime.strptime(str(value), "%Y-%m-%d").date()
+
+    def add_flag(code, *, weight, detail=None):
+        risk_flags.append(
+            {
+                "code": code,
+                "severity": "advisory",
+                "weight": weight,
+                "detail": detail or {},
+            }
+        )
+
+    def po_terms_tier():
+        po_number = getattr(extracted_fields, "po_number_canonical", None)
+
+        if not po_number:
+            return None
+
+        matching_pos = [
+            po
+            for po in purchase_orders
+            if po.tenant_id == document.tenant_id
+            and po.po_number_canonical == po_number
+            and getattr(po, "deleted_at", None) is None
+            and as_int(getattr(po, "payment_terms", None)) is not None
+        ]
+
+        if len(matching_pos) >= 1:
+            return {
+                "tier": "purchase_order",
+                "expected_payment_terms": as_int(matching_pos[0].payment_terms),
+                "source_id": getattr(matching_pos[0], "id", None),
+            }
+
+        return None
+
+    def contract_terms_tier():
+        invoice_date = as_date(getattr(extracted_fields, "invoice_date", None))
+
+        if not invoice_date:
+            skipped_steps.append(
+                {"step": "contract_payment_terms", "skip_reason": "no_invoice_date"}
+            )
+            return None
+
+        matching_contracts = []
+
+        for contract in contracts:
+            if contract.tenant_id != document.tenant_id:
+                continue
+
+            if contract.vendor_id != resolved_vendor_id:
+                continue
+
+            if contract.entity_id != document.entity_id:
+                continue
+
+            if getattr(contract, "status", None) != "active":
+                continue
+
+            if getattr(contract, "deleted_at", None) is not None:
+                continue
+
+            effective_from = as_date(getattr(contract, "effective_from", None))
+            effective_to = as_date(getattr(contract, "effective_to", None))
+
+            if effective_from is None or invoice_date < effective_from:
+                continue
+
+            if effective_to is not None and invoice_date > effective_to:
+                continue
+
+            matching_contracts.append(contract)
+
+        if len(matching_contracts) != 1:
+            return None
+
+        payment_terms = as_int(getattr(matching_contracts[0], "payment_terms", None))
+        if payment_terms is None:
+            return None
+
+        return {
+            "tier": "contract",
+            "expected_payment_terms": payment_terms,
+            "source_id": getattr(matching_contracts[0], "id", None),
+        }
+
+    def vendor_master_terms_tier():
+        payment_terms = as_int(getattr(vendor, "payment_terms", None))
+
+        if payment_terms is None:
+            return None
+
+        return {
+            "tier": "vendor_master",
+            "expected_payment_terms": payment_terms,
+            "source_id": getattr(vendor, "id", None),
+        }
+
+    risk_flags = []
+    advisory_flags = []
+    skipped_steps = []
+
+    invoice_payment_terms = as_int(getattr(extracted_fields, "payment_terms", None))
+
+    if invoice_payment_terms is None:
+        skipped_steps.append(
+            {
+                "step": "payment_terms_validation",
+                "skip_reason": "invoice_payment_terms_missing",
+            }
+        )
+
+        return {
+            "risk_flags": risk_flags,
+            "advisory_flags": advisory_flags,
+            "skipped_steps": skipped_steps,
+            "result": "skipped",
+            "expected_payment_terms": None,
+            "expected_source_tier": None,
+        }
+
+    resolved = (
+        po_terms_tier()
+        or contract_terms_tier()
+        or vendor_master_terms_tier()
+    )
+
+    if resolved is None:
+        add_flag(
+            "vendor_payment_terms_unconfigured",
+            weight=0.10,
+            detail={
+                "reason": "no_po_contract_or_vendor_payment_terms",
+            },
+        )
+
+        return {
+            "risk_flags": risk_flags,
+            "advisory_flags": advisory_flags,
+            "skipped_steps": skipped_steps,
+            "result": "unconfigured",
+            "expected_payment_terms": None,
+            "expected_source_tier": None,
+        }
+
+    expected_payment_terms = resolved["expected_payment_terms"]
+    expected_source_tier = resolved["tier"]
+
+    difference = abs(invoice_payment_terms - expected_payment_terms)
+
+    if difference > tolerance_days:
+        add_flag(
+            "payment_terms_mismatch",
+            weight=0.20,
+            detail={
+                "invoice_payment_terms": invoice_payment_terms,
+                "expected_payment_terms": expected_payment_terms,
+                "difference_days": difference,
+                "tolerance_days": tolerance_days,
+                "expected_source_tier": expected_source_tier,
+                "source_id": resolved.get("source_id"),
+            },
+        )
+
+        result = "mismatch"
+    else:
+        result = "matched"
+
+    return {
+        "risk_flags": risk_flags,
+        "advisory_flags": advisory_flags,
+        "skipped_steps": skipped_steps,
+        "result": result,
+        "invoice_payment_terms": invoice_payment_terms,
+        "expected_payment_terms": expected_payment_terms,
+        "expected_source_tier": expected_source_tier,
+        "tolerance_days": tolerance_days,
+    }

src/Invoice Validation Functions/platform_configs.json

@@ -0,0 +1,67 @@
+{
+    "_comment_vendor": "Vendor validation configs",
+
+    "validation.name_match_algorithm_version": "rapidfuzz.local.v1",
+
+    "validation.vendor_match_require_high_tier_contexts": [
+      "contract_match",
+      "bank_validation",
+      "approval_revalidation"
+    ],
+
+    "validation.rapidfuzz_high_threshold": 0.90,
+    "validation.rapidfuzz_medium_threshold": 0.75,
+    "validation.rapidfuzz_suggestion_threshold": 0.60,
+
+    "_comment_tax_id": "Tax ID validation configs",
+    "_comment_tax_id_handlers": "Built-in checksum handlers are implemented in Tax - ID validation.py (_BUILTIN_CHECKSUM_HANDLERS). They cover all families with a non-null checksum in the registry: gb_vat_mod97, de_vat_mod11, fr_vat_mod97, it_vat_luhn, es_nif, gstin_crc, abn_mod89, bn_luhn, sg_nric_fin. To override at runtime, supply a callable dict under validation.tax_id_checksum_handlers when building platform_configs.",
+
+    "validation.tax_id_family_registry": {
+        "VAT-GB":   {"pattern": "^(GB)?[0-9]{9}([0-9]{3})?$",                          "checksum": "gb_vat_mod97"},
+        "VAT-DE":   {"pattern": "^(DE)?[0-9]{9}$",                                      "checksum": "de_vat_mod11"},
+        "VAT-FR":   {"pattern": "^(FR)?[A-Z0-9]{2}[0-9]{9}$",                           "checksum": "fr_vat_mod97"},
+        "VAT-IT":   {"pattern": "^(IT)?[0-9]{11}$",                                     "checksum": "it_vat_luhn"},
+        "VAT-ES":   {"pattern": "^(ES)?[A-Z0-9][0-9]{7}[A-Z0-9]$",                     "checksum": "es_nif"},
+        "VAT-NL":   {"pattern": "^(NL)?[0-9]{9}B[0-9]{2}$",                            "checksum": null},
+        "GSTIN-IN": {"pattern": "^[0-9]{2}[A-Z]{5}[0-9]{4}[A-Z][1-9A-Z]Z[0-9A-Z]$",  "checksum": "gstin_crc"},
+        "PAN-IN":   {"pattern": "^[A-Z]{5}[0-9]{4}[A-Z]$",                             "checksum": null},
+        "EIN-US":   {"pattern": "^[0-9]{2}-?[0-9]{7}$",                                "checksum": null},
+        "ABN-AU":   {"pattern": "^[0-9]{11}$",                                          "checksum": "abn_mod89"},
+        "BN-CA":    {"pattern": "^[0-9]{9}([A-Z]{2}[0-9]{4})?$",                       "checksum": "bn_luhn"},
+        "TIN-SG":   {"pattern": "^(?:[STFG][0-9]{7}[A-Z]|[0-9]{9}[A-Z])$",            "checksum": "sg_nric_fin"}
+    },
+
+    "_comment_bank": "Bank validation configs",
+
+    "validation.iban_checksum_required": true,
+    "validation.iban_accno_substring_exempt_countries": ["LC", "MT"],
+
+    "_comment_date": "Date sanity configs",
+    "validation.default_timezone": "Asia/Kolkata",
+    "validation.max_invoice_age_days": 365,
+    "validation.service_end_post_invoice_grace_days": 0,
+
+    "_comment_payment": "Payment terms configs",
+    "validation.payment_terms_tolerance_days": 2,
+
+    "_comment_tax": "Tax configs",
+    "validation.tax_rate_match_tolerance": "0.0001",
+    "iso_currencies.minor_unit_value": {
+        "USD": "0.01",
+        "GBP": "0.01",
+        "EUR": "0.01",
+        "INR": "0.01",
+        "JPY": "1",
+        "BHD": "0.001"
+    },
+    "validation.eu_member_states": [
+        "AT", "BE", "BG", "HR", "CY", "CZ", "DK", "EE", "FI", "FR",
+        "DE", "GR", "HU", "IE", "IT", "LV", "LT", "LU", "MT", "NL",
+        "PL", "PT", "RO", "SK", "SI", "ES", "SE"
+    ],
+
+    "_comment_completeness": "Completeness configs",
+    "validation.solver_tolerance.cap_minor_units": 5,
+    "validation.solver_tolerance.base_per_minor_unit": 1,
+    "validation.solver_tolerance.per_line_minor_units": 1
+}

src/Invoice Validation Functions/tax validation.py

@@ -0,0 +1,563 @@
+from decimal import Decimal, ROUND_HALF_UP
+from datetime import date, datetime
+
+
+def run_tax_amount_validation(
+    *,
+    document,
+    extracted_fields,
+    line_items,
+    vendor,
+    entity,
+    tax_master_rows,
+    platform_configs=None,
+):
+    """
+    Tax percentage and tax amount validation.
+
+    Advisory checks:
+      - tax rate exists in tax_master
+      - tax amount matches taxable base
+      - cross-country routing
+      - US sales/use tax advisories
+      - UK VAT advisories
+      - India GST advisories
+      - EU reverse charge
+      - vendor withholding advisory
+      - vendor tax exemption handling
+    """
+
+    platform_configs = platform_configs or {}
+
+    tax_rate_tolerance = Decimal(
+        str(platform_configs.get("validation.tax_rate_match_tolerance", "0.0001"))
+    )
+
+    currency_minor_units = platform_configs.get(
+        "iso_currencies.minor_unit_value",
+        {
+            "USD": Decimal("0.01"),
+            "GBP": Decimal("0.01"),
+            "EUR": Decimal("0.01"),
+            "INR": Decimal("0.01"),
+            "JPY": Decimal("1"),
+            "BHD": Decimal("0.001"),
+        },
+    )
+
+    eu_member_states = set(
+        platform_configs.get(
+            "validation.eu_member_states",
+            [
+                "AT", "BE", "BG", "HR", "CY", "CZ", "DK", "EE", "FI", "FR",
+                "DE", "GR", "HU", "IE", "IT", "LV", "LT", "LU", "MT", "NL",
+                "PL", "PT", "RO", "SK", "SI", "ES", "SE",
+            ],
+        )
+    )
+
+    def as_decimal(value):
+        if value is None or value == "":
+            return None
+        try:
+            return Decimal(str(value))
+        except Exception:
+            return None
+
+    def country(value):
+        if not value:
+            return None
+        return str(value).upper()
+
+    def is_zero(value):
+        value = as_decimal(value)
+        return value is not None and value == Decimal("0")
+
+    def is_non_zero(value):
+        value = as_decimal(value)
+        return value is not None and value != Decimal("0")
+
+    def invoice_date_value():
+        value = getattr(extracted_fields, "invoice_date", None)
+        if isinstance(value, datetime):
+            return value.date()
+        if isinstance(value, date):
+            return value
+        return None  # do not invent today's date
+
+    def currency_tolerance():
+        invoice_currency = country(getattr(extracted_fields, "currency", None))
+        raw = currency_minor_units.get(invoice_currency)
+        return Decimal(str(raw)) if raw is not None else None
+
+    def add_flag(code, *, weight=None, detail=None):
+        flag = {"code": code, "severity": "advisory"}
+        if weight is not None:
+            flag["weight"] = weight
+        if detail is not None:
+            flag["detail"] = detail
+        risk_flags.append(flag)
+
+    def add_advisory(code, *, detail=None):
+        advisory = {"code": code, "severity": "advisory"}
+        if detail is not None:
+            advisory["detail"] = detail
+        # avoid duplicates
+        if not any(a["code"] == code for a in advisory_flags):
+            advisory_flags.append(advisory)
+
+    def add_skip(step_name, reason):
+        skipped_steps.append({"step": step_name, "skip_reason": reason})
+
+    def effective_tax_master_rows():
+        invoice_date = invoice_date_value()
+        rows = []
+        for row in tax_master_rows:
+            if getattr(row, "tenant_id", None) != document.tenant_id:
+                continue
+            if not getattr(row, "is_active", True):
+                continue
+            if getattr(row, "deleted_at", None) is not None:
+                continue
+
+            # If invoice date is unavailable, do not invent one.
+            # Keep only active rows, but skip date-sensitive matching later if needed.
+            if invoice_date is not None:
+                effective_from = getattr(row, "effective_from", None)
+                effective_to = getattr(row, "effective_to", None)
+
+                if effective_from is not None and effective_from > invoice_date:
+                    continue
+                if effective_to is not None and invoice_date >= effective_to:
+                    continue
+
+            rows.append(row)
+        return rows
+
+    def rate_matches(master_rate, invoice_rate):
+        master_rate = as_decimal(master_rate)
+        invoice_rate = as_decimal(invoice_rate)
+        if master_rate is None or invoice_rate is None:
+            return False
+        return abs(master_rate - invoice_rate) <= tax_rate_tolerance
+
+    def row_region_matches(row_region, hint_region):
+        if row_region is None:
+            return True
+        return hint_region is not None and str(row_region).upper() == str(hint_region).upper()
+
+    def find_tax_master_rate(
+        *,
+        rate,
+        country_code,
+        region_code=None,
+        allowed_tax_types=None,
+        reduced_rate_allowed=False,
+    ):
+        country_code = country(country_code)
+        matches = []
+
+        for row in active_tax_master:
+            if country(getattr(row, "country_code", None)) != country_code:
+                continue
+
+            if not row_region_matches(getattr(row, "region_code", None), region_code):
+                continue
+
+            if allowed_tax_types is not None and getattr(row, "tax_type", None) not in allowed_tax_types:
+                continue
+
+            if rate_matches(getattr(row, "tax_rate", None), rate):
+                matches.append(row)
+                continue
+
+            elif reduced_rate_allowed:
+                tax_name = str(getattr(row, "tax_name", "") or "").lower()
+                if "reduced" in tax_name:
+                    matches.append(row)
+
+        return matches
+
+    def find_india_gst_aggregate_rate(rate, region_code, jurisdiction):
+        """
+        Match invoice aggregate GST rate against the sum of applicable component rates:
+          inter_state : IGST single-row match
+          intra_state : CGST + SGST pair whose rates sum to the invoice rate
+          None/unknown: try IGST, then CGST+SGST, then CGST+UTGST
+        """
+
+        def in_rows(tax_type, rgn=None):
+            return [
+                r for r in active_tax_master
+                if country(getattr(r, "country_code", None)) == "IN"
+                and getattr(r, "tax_type", None) == tax_type
+                and row_region_matches(getattr(r, "region_code", None), rgn)
+            ]
+
+        def sum_pair(type_a, type_b, rgn=None):
+            for a in in_rows(type_a, rgn):
+                a_rate = as_decimal(getattr(a, "tax_rate", None))
+                if a_rate is None:
+                    continue
+                for b in in_rows(type_b, rgn):
+                    b_rate = as_decimal(getattr(b, "tax_rate", None))
+                    if b_rate is None:
+                        continue
+                    if abs((a_rate + b_rate) - rate) <= tax_rate_tolerance:
+                        return [a, b]
+            return []
+
+        if jurisdiction == "inter_state":
+            for r in in_rows("IGST", region_code):
+                if rate_matches(getattr(r, "tax_rate", None), rate):
+                    return [r]
+            return []
+
+        if jurisdiction == "intra_state":
+            return sum_pair("CGST", "SGST", region_code)
+
+        # Unknown jurisdiction: try all applicable component combinations.
+        for r in in_rows("IGST"):
+            if rate_matches(getattr(r, "tax_rate", None), rate):
+                return [r]
+        result = sum_pair("CGST", "SGST")
+        if result:
+            return result
+        return sum_pair("CGST", "UTGST")
+
+    def all_line_rates_zero():
+        for line in line_items:
+            rate = as_decimal(getattr(line, "tax_rate_per_item", None))
+            if rate is None:
+                continue
+            if rate != Decimal("0"):
+                return False
+        return True
+
+    def sender_tax_family():
+        return getattr(extracted_fields, "sender_tax_id_inferred_family", None)
+
+    def sender_tax_valid_for_country(expected_family):
+        return (
+            getattr(extracted_fields, "sender_tax_id", None) is not None
+            and sender_tax_family() == expected_family
+            and bool(getattr(extracted_fields, "sender_tax_id_format_valid", True))
+            and bool(getattr(extracted_fields, "sender_tax_id_checksum_valid", True))
+        )
+
+    def supplier_and_recipient_are_eu():
+        return supplier_country in eu_member_states and recipient_country in eu_member_states
+
+    def eu_reverse_charge_conditions_met():
+        if not supplier_and_recipient_are_eu():
+            return False
+        if supplier_country == recipient_country:
+            return False
+
+        if vendor_exemption_status == "reverse_charge":
+            return True
+
+        supplier_family = f"VAT-{supplier_country}"
+        supplier_vat_valid = sender_tax_valid_for_country(supplier_family)
+        recipient_vat_present = bool(getattr(entity, "vat_id", None))
+
+        return (
+            supplier_vat_valid
+            and recipient_vat_present
+            and all_line_rates_zero()
+            and is_zero(getattr(extracted_fields, "tax_amount", None))
+        )
+
+    def india_jurisdiction():
+        supplier_state = getattr(vendor, "billing_state", None)
+        recipient_state = getattr(entity, "region_code", None)
+        if not supplier_state or not recipient_state:
+            return None
+        if str(supplier_state).upper() == str(recipient_state).upper():
+            return "intra_state"
+        return "inter_state"
+
+    risk_flags = []
+    advisory_flags = []
+    skipped_steps = []
+    ops_alerts = []
+    matched_tax_master_rows = {}
+
+    active_tax_master = effective_tax_master_rows()
+
+    supplier_country = country(getattr(vendor, "billing_country", None))
+    recipient_country = country(getattr(entity, "country_code", None))
+    invoice_currency = country(getattr(extracted_fields, "currency", None))
+    vendor_exemption_status = getattr(vendor, "exemption_status", None)
+    invoice_date = invoice_date_value()
+
+    if not active_tax_master:
+        ops_alerts.append(
+            {
+                "code": "alert.tax_master_unconfigured",
+                "tenant_id": document.tenant_id,
+            }
+        )
+        return {
+            "risk_flags": risk_flags,
+            "advisory_flags": advisory_flags,
+            "skipped_steps": [
+                {"step": "all_tax_checks", "skip_reason": "tax_master_unconfigured"}
+            ],
+            "ops_alerts": ops_alerts,
+            "matched_tax_master_rows": matched_tax_master_rows,
+        }
+
+    # Check 1 and Check 2 both skip if supplier country is missing.
+    if not supplier_country:
+        add_skip("rate_exists_in_master", "supplier_country_missing")
+        add_skip("tax_amount_matches_taxable_base", "supplier_country_missing")
+    else:
+        unique_rates = set()
+
+        # Gather unique (rate, region) pairs — deduplicated by value per SDD §8.1.
+        for line in line_items:
+            rate = as_decimal(getattr(line, "tax_rate_per_item", None))
+            if rate is not None and rate != Decimal("0"):
+                unique_rates.add(
+                    (
+                        rate,
+                        getattr(line, "region_code_hint", None),
+                    )
+                )
+
+        # Summary tax rate is checked separately
+        summary_rate = as_decimal(getattr(extracted_fields, "tax_rate", None))
+        if summary_rate is not None and summary_rate != Decimal("0"):
+            unique_rates.add((summary_rate, None))
+
+        for rate, region_code in unique_rates:
+            # India aggregate rate matching is handled entirely in Check 6.
+            if supplier_country == "IN":
+                continue
+
+            # US state tax needs a region hint
+            if supplier_country == "US" and not region_code:
+                add_advisory("us_region_unresolved")
+                continue
+
+            allowed_tax_types = None
+            if supplier_country == "GB":
+                allowed_tax_types = {"VAT"}
+
+            reduced_rate_allowed = vendor_exemption_status == "reduced_rate"
+
+            matches = find_tax_master_rate(
+                rate=rate,
+                country_code=supplier_country,
+                region_code=region_code,
+                allowed_tax_types=allowed_tax_types,
+                reduced_rate_allowed=reduced_rate_allowed,
+            )
+
+            if matches:
+                matched_tax_master_rows[str(rate)] = [getattr(row, "id", None) for row in matches]
+            else:
+                add_flag(
+                    "tax_rate_not_in_master",
+                    weight=Decimal("0.35"),
+                    detail={
+                        "rate": str(rate),
+                        "country": supplier_country,
+                        "region_code": region_code,
+                    },
+                )
+
+
+    # Check 2: Amount matches taxable base.
+    if supplier_country:
+        tolerance = currency_tolerance()
+        if tolerance is None:
+            add_skip("tax_amount_matches_taxable_base", "currency_minor_unit_unknown")
+        else:
+            for line in line_items:
+                rate = as_decimal(getattr(line, "tax_rate_per_item", None))
+                amount = as_decimal(getattr(line, "amount", None))
+                actual_tax = as_decimal(getattr(line, "tax_amount_per_item", None))
+
+                if rate is None or amount is None or actual_tax is None:
+                    continue
+
+                discount = as_decimal(getattr(line, "discount_amount_per_item", None)) or Decimal("0")
+                taxable_base = amount - discount
+                expected_tax = taxable_base * (rate / Decimal("100"))
+
+                if abs(expected_tax - actual_tax) > tolerance:
+                    add_flag(
+                        "tax_amount_mismatch",
+                        weight=Decimal("0.35"),
+                        detail={
+                            "line_number": getattr(line, "line_number", None),
+                            "expected_tax": str(expected_tax.quantize(tolerance, rounding=ROUND_HALF_UP)),
+                            "actual_tax": str(actual_tax),
+                            "currency": invoice_currency,
+                        },
+                    )
+
+    # §8.9: vendor_tax_exempt advisory fires when all extracted line rates are zero.
+    if vendor_exemption_status == "exempt" and all_line_rates_zero():
+        add_advisory("vendor_tax_exempt")
+
+    # Check 3: Cross-country routing.
+    summary_tax_amount = as_decimal(getattr(extracted_fields, "tax_amount", None))
+
+    if supplier_country is None or summary_tax_amount is None or summary_tax_amount == Decimal("0"):
+        add_skip("cross_country_routing", "supplier_country_or_tax_amount_missing_or_zero")
+    elif recipient_country is None:
+        add_skip("cross_country_routing", "recipient_country_missing")
+    elif supplier_country != recipient_country:
+        add_flag(
+            "cross_country_tax",
+            weight=Decimal("0.30"),
+            detail={
+                "supplier_country": supplier_country,
+                "recipient_country": recipient_country,
+            },
+        )
+
+    # Check 4: United States.
+    if supplier_country == "US" or recipient_country == "US":
+        if supplier_country == "US" and summary_tax_amount is not None and summary_tax_amount != Decimal("0"):
+            add_flag(
+                "us_sales_tax",
+                weight=Decimal("0.20"),
+                detail={"tax_amount": str(summary_tax_amount)},
+            )
+
+        if (
+            recipient_country == "US"
+            and supplier_country != recipient_country
+            and all_line_rates_zero()
+        ):
+            add_flag(
+                "us_use_tax_possibly_owed",
+                weight=Decimal("0.15"),
+            )
+
+    # Check 5: United Kingdom.
+    if supplier_country == "GB" or recipient_country == "GB":
+        for line in line_items:
+            rate = as_decimal(getattr(line, "tax_rate_per_item", None))
+            tax_amount = as_decimal(getattr(line, "tax_amount_per_item", None))
+
+            if rate == Decimal("0"):
+                gb_matches = find_tax_master_rate(
+                    rate=rate,
+                    country_code="GB",
+                    region_code=getattr(line, "region_code_hint", None),
+                    allowed_tax_types={"zero_rated", "exempt", "VAT"},
+                    reduced_rate_allowed=False,
+                )
+
+                if any(getattr(row, "tax_type", None) == "zero_rated" for row in gb_matches):
+                    add_advisory("uk_zero_rated", detail={"line_number": getattr(line, "line_number", None)})
+                elif tax_amount == Decimal("0") and any(
+                    getattr(row, "tax_type", None) == "exempt" for row in gb_matches
+                ):
+                    add_advisory("uk_exempt", detail={"line_number": getattr(line, "line_number", None)})
+                elif not (
+                    vendor_exemption_status == "reverse_charge"
+                    and supplier_and_recipient_are_eu()
+                    and supplier_country != recipient_country
+                ) and not (
+                    vendor_exemption_status == "exempt"
+                    and all_line_rates_zero()
+                ):
+                    add_flag(
+                        "tax_rate_not_in_master",
+                        weight=Decimal("0.35"),
+                        detail={
+                            "rate": "0",
+                            "country": "GB",
+                            "source": getattr(line, "line_number", None),
+                        },
+                    )
+
+        if (
+            supplier_country == "GB"
+            and recipient_country == "GB"
+            and summary_tax_amount is not None
+            and summary_tax_amount != Decimal("0")
+            and not sender_tax_valid_for_country("VAT-GB")
+        ):
+            add_flag(
+                "missing_tax_id_for_local_tax",
+                weight=Decimal("0.25"),
+                detail={"expected_family": "VAT-GB"},
+            )
+
+    # Check 6: India GST.
+    if supplier_country == "IN" or recipient_country == "IN":
+        jurisdiction = india_jurisdiction()
+        if jurisdiction is None:
+            add_advisory("in_gst_jurisdiction_unresolved")
+        else:
+            add_advisory("in_gst_jurisdiction_inferred", detail={"jurisdiction": jurisdiction})
+
+        # Sender tax ID must look like GSTIN-IN or PAN-IN
+        if sender_tax_family() not in {"GSTIN-IN", "PAN-IN"}:
+            add_flag(
+                "missing_tax_id_for_local_tax",
+                weight=Decimal("0.25"),
+                detail={"expected_family": "GSTIN-IN or PAN-IN"},
+            )
+
+        for line in line_items:
+            rate = as_decimal(getattr(line, "tax_rate_per_item", None))
+            if rate is None or rate == Decimal("0"):
+                continue
+
+            region_code = None
+            if jurisdiction == "intra_state":
+                region_code = getattr(vendor, "billing_state", None) or getattr(entity, "region_code", None)
+
+            matches = find_india_gst_aggregate_rate(rate, region_code, jurisdiction)
+
+            if matches:
+                matched_tax_master_rows[getattr(line, "line_number", None)] = [
+                    getattr(r, "id", None) for r in matches
+                ]
+            else:
+                add_flag(
+                    "tax_rate_not_in_master",
+                    weight=Decimal("0.35"),
+                    detail={
+                        "rate": str(rate),
+                        "country": "IN",
+                        "region_code": region_code,
+                        "source": getattr(line, "line_number", None),
+                    },
+                )
+
+    # Check 7: EU reverse charge.
+    # If conditions are met, add advisory and retract cross_country_tax raised by Check 3.
+    if supplier_and_recipient_are_eu() and supplier_country != recipient_country:
+        if eu_reverse_charge_conditions_met():
+            add_advisory("eu_reverse_charge")
+            risk_flags[:] = [f for f in risk_flags if f["code"] != "cross_country_tax"]
+
+    # Check 8: Withholding tax advisory.
+    if vendor_exemption_status == "withholding_applicable":
+        add_advisory("vendor_withholding_applicable")
+
+    for line in line_items:
+        tax_master_id = getattr(line, "tax_master_id", None)
+        if not tax_master_id:
+            continue
+        for row in active_tax_master:
+            if getattr(row, "id", None) == tax_master_id and getattr(row, "tax_type", None) == "withholding":
+                add_advisory("vendor_withholding_applicable")
+                break
+
+    return {
+        "risk_flags": risk_flags,
+        "advisory_flags": advisory_flags,
+        "skipped_steps": skipped_steps,
+        "ops_alerts": ops_alerts,
+        "matched_tax_master_rows": matched_tax_master_rows,
+    }