State Charity Registration Compliance

This module is part of the Core Architecture & Compliance Mapping reference, where it operates as the discrete subsystem that tracks multi-jurisdictional charitable-solicitation registration, schedules statutory renewals, and validates state filing obligations. Its mandate is narrow and explicit: accept validated organizational metadata, evaluate each entity’s state-specific registration obligations, and emit deterministic, audit-ready compliance artifacts. It accepts already-typed payloads and it terminates at the filing-manifest handoff.

Explicitly out of scope. This subsystem does not parse source documents, disburse grant funds, or authenticate to state submission portals. Document-level extraction — PDF coordinate mapping and OCR confidence scoring — belongs to the Data Ingestion & Grant Parsing Workflows reference. Canonical tax-exempt financial fields arrive pre-normalized from the IRS 990 Data Schema Mapping layer; this module only consumes them. Funder-eligibility decisions are owned by Grantor-Specific Rule Taxonomies, encryption and access control by Data Security & Access Boundaries, and retry orchestration by Pipeline Fallback & Retry Logic. By enforcing this separation, the registration layer guarantees that every renewal window and filing manifest it emits is reproducible, lineage-tagged, and free of cross-stage state mutation.

Four stakeholders depend on its outputs: nonprofit operations teams (jurisdictional intake and manual exception resolution), grant managers (renewal exposure windows tied to funding eligibility), Python automation developers (validation schemas and deterministic routing logic), and compliance officers (immutable execution logs and statutory filing approval).

Prerequisites #

The pipeline pins its runtime and dependencies so that obligation evaluation is byte-for-byte reproducible across CI and production workers — a renewal window computed today must match the window recomputed during an audit two years later.

• Python: 3.11 or newer (relies on datetime.UTC and timezone-aware date arithmetic).
• Pinned packages:

# requirements.txt (state-registration layer only)
pydantic==2.7.1          # canonical ingestion validation models
rapidfuzz==3.9.0         # deterministic EIN fuzzy reconciliation
python-dateutil==2.9.0   # tolerant fiscal-year date parsing
pdfplumber==0.11.0       # bounded-box text extraction for scanned notices
pytest==8.2.0            # validation, scheduling, and ledger tests
hypothesis==6.100.1      # property-based renewal-window edge cases

• Environment variables: REG_QUARANTINE_SINK (write target for rejected payloads), REG_FUZZY_THRESHOLD (EIN match floor, default 0.92), REG_CRITICAL_DAYS (exposure threshold, default 30), and AUDIT_SINK_URL (write-once ledger target).
• Upstream dependency: this stage consumes the validated, type-safe entity record produced by the IRS 990 Data Schema Mapping layer and the parsed dictionaries from Field Mapping & Normalization. It assumes transport-level extraction has already succeeded; it never re-opens source files.

Core Implementation #

The registration engine is a five-stage pipeline. Each stage terminates with an explicit handoff contract: downstream consumers receive only serialized, hash-verifiable payloads, and no stage may mutate the state of another. The four code blocks below implement the load-bearing stages — ingestion validation, entity reconciliation, obligation evaluation, and filing validation — plus the audit ledger that binds them.

Stage I — Ingestion & strict validation #

Raw state registration documents, exemption certificates, and renewal notices enter only through authenticated endpoints. PDF notices are text-extracted with pdfplumber using explicit bounding boxes to avoid header/footer contamination. Every record is then forced through a Pydantic model; anything that fails type coercion or carries a non-canonical jurisdiction code is routed to a deterministic quarantine queue with a structured rejection payload rather than silently dropped.

import hashlib
import logging
from datetime import datetime, timezone
from pydantic import BaseModel, Field, field_validator
from pydantic import ValidationError as PydanticValidationError

logger = logging.getLogger("registration.ingestion")

VALID_JURISDICTIONS = {
    "AL", "AK", "AZ", "AR", "CA", "CO", "CT", "DE", "FL", "GA", "HI", "ID", "IL",
    "IN", "IA", "KS", "KY", "LA", "ME", "MD", "MA", "MI", "MN", "MS", "MO", "MT",
    "NE", "NV", "NH", "NJ", "NM", "NY", "NC", "ND", "OH", "OK", "OR", "PA", "RI",
    "SC", "SD", "TN", "TX", "UT", "VT", "VA", "WA", "WV", "WI", "WY", "DC",
}


class RegistrationIngestionPayload(BaseModel):
    ein: str = Field(pattern=r"^\d{2}-\d{7}$")
    state_registration_id: str
    jurisdiction_code: str = Field(pattern=r"^[A-Z]{2}$")
    registration_status: str = Field(pattern=r"^(Active|Pending|Suspended|Revoked|Expired)$")
    expiration_date: datetime
    filing_type: str

    @field_validator("jurisdiction_code")
    @classmethod
    def validate_iso_jurisdiction(cls, v: str) -> str:
        if v not in VALID_JURISDICTIONS:
            raise ValueError(f"Non-canonical ISO 3166-2 jurisdiction code: {v}")
        return v


def process_ingestion(raw_payload: dict, source_system: str) -> dict:
    payload_hash = hashlib.sha256(repr(raw_payload).encode("utf-8")).hexdigest()
    audit_ctx = {
        "ingestion_id": payload_hash[:12],
        "timestamp_utc": datetime.now(timezone.utc).isoformat(),
        "source_system": source_system,
    }
    try:
        validated = RegistrationIngestionPayload(**raw_payload)
        audit_ctx["validation_result"] = "PASS"
        logger.info("ingestion.validated", extra=audit_ctx)
        return {"status": "VALIDATED", "payload": validated.model_dump(mode="json"), "audit": audit_ctx}
    except PydanticValidationError as exc:
        first = exc.errors()[0] if exc.errors() else {"loc": ("unknown",)}
        field_path = first["loc"][0]
        rejection = {
            "error_code": "SCHEMA_VALIDATION_FAILURE",
            "field_path": field_path,
            "raw_value": raw_payload.get(field_path, "null"),
            "audit": audit_ctx,
        }
        logger.warning("ingestion.quarantined", extra=rejection)
        return {"status": "QUARANTINED", "rejection": rejection}

Every ingestion attempt — pass or quarantine — appends an immutable entry to the write-once ledger before the payload advances. Encryption-at-rest (AES-256-GCM) and least-privilege IAM are enforced at this edge per Data Security & Access Boundaries; PII and financial fields are masked in transit logs.

Stage II — Reconciliation & canonical mapping #

Validated payloads are matched against the internal master entity registry. Exact ein + jurisdiction_code joins are preferred; legacy EIN formatting variants fall back to a rapidfuzz similarity score gated at the REG_FUZZY_THRESHOLD floor. No rule evaluation happens here — reconciliation only resolves identity and emits a confidence-scored lineage tag.

import logging
from rapidfuzz import fuzz

logger = logging.getLogger("registration.reconcile")

MASTER_REGISTRY: list[dict[str, str]] = [
    {"ein": "12-3456789", "jurisdiction_code": "NY", "canonical_entity_id": "ENT-001"},
    {"ein": "98-7654321", "jurisdiction_code": "TX", "canonical_entity_id": "ENT-002"},
]


def reconcile_entity(payload: dict, fuzzy_threshold: float = 0.92) -> dict:
    ein = payload["ein"]
    jurisdiction = payload["jurisdiction_code"]

    for record in MASTER_REGISTRY:
        if record["ein"] == ein and record["jurisdiction_code"] == jurisdiction:
            logger.info("reconcile.exact", extra={"ein": ein, "jurisdiction": jurisdiction})
            return {
                "canonical_entity_id": record["canonical_entity_id"],
                "resolved_jurisdiction": jurisdiction,
                "match_type": "EXACT",
                "confidence_score": 1.0,
            }

    best = max(MASTER_REGISTRY, key=lambda r: fuzz.ratio(r["ein"], ein))
    score = fuzz.ratio(best["ein"], ein) / 100.0
    if score >= fuzzy_threshold:
        logger.info("reconcile.fuzzy", extra={"ein": ein, "score": round(score, 3)})
        return {
            "canonical_entity_id": best["canonical_entity_id"],
            "resolved_jurisdiction": jurisdiction,
            "match_type": "FUZZY_FALLBACK",
            "confidence_score": round(score, 3),
        }

    logger.warning("reconcile.unresolved", extra={"ein": ein, "jurisdiction": jurisdiction})
    return {"match_type": "UNRESOLVED", "confidence_score": 0.0}

Unresolved entities raise an exception ticket routed to Nonprofit Operations for manual registry reconciliation rather than proceeding on a guess.

Stage III — Obligation evaluation & renewal scheduling #

Reconciled entities are evaluated against jurisdiction-specific filing calendars. Exposure is a pure function of days-to-expiration, so identical inputs produce identical scheduling outputs regardless of when the function runs — the idempotency property auditors rely on.

import logging
from datetime import date, datetime, timedelta

logger = logging.getLogger("registration.obligation")


def evaluate_renewal_obligation(
    expiration_date_iso: str, jurisdiction: str, critical_days: int = 30
) -> dict:
    expiration = datetime.fromisoformat(expiration_date_iso).date()
    days_remaining = (expiration - date.today()).days

    if days_remaining <= 0:
        status, exposure = "EXPIRED", "CRITICAL"
    elif days_remaining <= critical_days:
        status, exposure = "ACTIVE", "CRITICAL"
    elif days_remaining <= 90:
        status, exposure = "ACTIVE", "WARNING"
    else:
        status, exposure = "ACTIVE", "STABLE"

    obligation = {
        "canonical_status": status,
        "exposure_level": exposure,
        "days_remaining": days_remaining,
        "filing_window": {
            "start_iso": (expiration - timedelta(days=60)).isoformat(),
            "end_iso": (expiration - timedelta(days=7)).isoformat(),
        },
        "jurisdiction": jurisdiction,
    }
    logger.info("obligation.scheduled", extra={"jurisdiction": jurisdiction, "exposure": exposure})
    return obligation

Exposure flags map directly into Grantor-Specific Rule Taxonomies so that no disbursement reaches an entity in a lapsed or CRITICAL registration state.

Stage IV — Filing validation & execution routing #

Scheduled obligations are cross-referenced against the authoritative state reporting matrix, required attachments are checked, and a deterministic filing manifest is emitted. Routing is fire-and-forget — portal authentication and submission retries are delegated to the dedicated filing microservice governed by Pipeline Fallback & Retry Logic.

import logging

logger = logging.getLogger("registration.filing")

STATE_FILING_MATRIX = {
    "NY": {"form": "CHAR500", "requires_990": True, "fee_tier": "A"},
    "CA": {"form": "RRF-1", "requires_990": True, "fee_tier": "B"},
    "TX": {"form": "802", "requires_990": False, "fee_tier": "C"},
}


def validate_filing_manifest(jurisdiction: str, has_990: bool) -> dict:
    reqs = STATE_FILING_MATRIX.get(jurisdiction)
    if reqs is None:
        logger.warning("filing.unsupported", extra={"jurisdiction": jurisdiction})
        return {"status": "REJECTED", "reason": "UNSUPPORTED_JURISDICTION"}

    missing = ["IRS_FORM_990"] if reqs["requires_990"] and not has_990 else []
    if missing:
        return {
            "status": "INCOMPLETE",
            "missing_attachments": missing,
            "remediation": f"Upload {', '.join(missing)} before routing to execution.",
        }

    return {
        "status": "READY_FOR_EXECUTION",
        "manifest": {
            "jurisdiction": jurisdiction,
            "form_id": reqs["form"],
            "fee_tier": reqs["fee_tier"],
            "submission_protocol": "HTTPS_MTLS",
        },
    }

Manifest construction adheres strictly to the State-by-state nonprofit reporting requirements checklist, which is the version-controlled source of truth for forms, thresholds, and fee tiers.

Stage V — Immutable audit ledger #

All four stage outputs converge into an append-only ledger that chains SHA-256 hashes into a Merkle-style verification trail. Any correction is a compensating entry with explicit correction_of lineage — entries are never edited in place.

import hashlib
import json
import logging

logger = logging.getLogger("registration.audit")


class AuditLedger:
    def __init__(self) -> None:
        self.entries: list[dict] = []
        self.chain_hash: str = "0" * 64  # genesis hash

    def append_entry(self, stage: str, payload: dict, timestamp: str) -> str:
        entry_data = json.dumps(
            {"stage": stage, "payload": payload, "timestamp": timestamp}, sort_keys=True
        )
        current_hash = hashlib.sha256(f"{self.chain_hash}{entry_data}".encode()).hexdigest()
        self.entries.append(
            {"hash": current_hash, "prev_hash": self.chain_hash, "data": json.loads(entry_data)}
        )
        self.chain_hash = current_hash
        logger.info("audit.appended", extra={"stage": stage, "hash": current_hash[:12]})
        return current_hash

    def export_ledger(self) -> dict:
        return {
            "total_entries": len(self.entries),
            "terminal_hash": self.chain_hash,
            "entries": self.entries,
        }

The ledger’s terminal_hash is the single value an examiner verifies to confirm that no stage output was altered after the fact. Its metadata conforms to Compliance Metadata Standards.

Field Mapping & Schema Contract #

State source systems and legacy spreadsheets describe the same concepts with inconsistent names. The reconciliation stage collapses every alias into one canonical key before any rule runs, and coerces each value to a strict type. Records that cannot be coerced are quarantined, never silently nulled.

Jurisdiction-alias resolution is the highest-volume source of quarantine, so it is centralized in one lookup table:

JURISDICTION_ALIASES: dict[str, str] = {
    "california": "CA", "calif.": "CA", "cal": "CA",
    "new york": "NY", "n.y.": "NY",
    "texas": "TX", "tex.": "TX",
}


def canonicalize_jurisdiction(raw: str) -> str | None:
    token = raw.strip().lower()
    if len(token) == 2 and token.upper() in VALID_JURISDICTIONS:
        return token.upper()
    return JURISDICTION_ALIASES.get(token)

Validation & Testing #

Because exposure classification and filing-window math drive funding decisions, every boundary is locked with pytest and probed with hypothesis. Tests assert both the deterministic output and the audit-log side effect.

import logging
from datetime import date, timedelta
from hypothesis import given, strategies as st


def test_critical_within_thirty_days() -> None:
    expiry = (date.today() + timedelta(days=15)).isoformat()
    result = evaluate_renewal_obligation(expiry, "CA")
    assert result["exposure_level"] == "CRITICAL"
    assert result["canonical_status"] == "ACTIVE"


def test_expired_is_critical() -> None:
    expiry = (date.today() - timedelta(days=1)).isoformat()
    result = evaluate_renewal_obligation(expiry, "NY")
    assert result["exposure_level"] == "CRITICAL"
    assert result["canonical_status"] == "EXPIRED"


def test_missing_990_blocks_ny_filing() -> None:
    result = validate_filing_manifest("NY", has_990=False)
    assert result["status"] == "INCOMPLETE"
    assert "IRS_FORM_990" in result["missing_attachments"]


def test_bad_jurisdiction_quarantines(caplog) -> None:
    bad = {
        "ein": "12-3456789", "state_registration_id": "X1",
        "jurisdiction_code": "ZZ", "registration_status": "Active",
        "expiration_date": "2027-01-01T00:00:00", "filing_type": "ANNUAL",
    }
    with caplog.at_level(logging.WARNING):
        out = process_ingestion(bad, source_system="unit-test")
    assert out["status"] == "QUARANTINED"
    assert out["rejection"]["error_code"] == "SCHEMA_VALIDATION_FAILURE"
    assert "ingestion.quarantined" in caplog.text


@given(days=st.integers(min_value=91, max_value=3650))
def test_stable_band_is_monotonic(days: int) -> None:
    expiry = (date.today() + timedelta(days=days)).isoformat()
    assert evaluate_renewal_obligation(expiry, "TX")["exposure_level"] == "STABLE"

A passing payload returns {"status": "VALIDATED", ...} with an ingestion_id; a failing one returns {"status": "QUARANTINED", ...} and emits exactly one ingestion.quarantined warning. The property test guarantees that any expiration beyond 90 days resolves to STABLE — no off-by-one at the band edges.

Performance & Scale Considerations #

Nonprofit-scale registration workloads are bursty but small by data-engineering standards: a multi-state charity tracks tens to low-hundreds of active registrations, and even a fiscal-sponsor managing hundreds of sub-entities rarely exceeds a few hundred thousand records.

• Batch sizing. Process ingestion in batches of 500 payloads. The Pydantic validation cost is dominated by regex compilation, which is amortized once the model class is loaded, so larger batches yield diminishing returns and raise quarantine-replay cost.
• Reconciliation cost. Exact matching is O(1) against an indexed registry; the rapidfuzz fallback is O(n) over candidate EINs. Cap the fuzzy candidate set per record and short-circuit on the first exact hit to keep the fallback off the hot path.
• Concurrency limits. Stages are pure functions over serialized payloads, so they parallelize cleanly. Bound worker pools to 8–16 concurrent ingestion workers; the practical ceiling is the write-once ledger’s append throughput, not CPU.
• Memory ceilings. Stream PDF extraction page-by-page with pdfplumber and release page objects immediately — a scanned multi-state renewal packet otherwise pins tens of MB per worker. Keep the master registry resident (it is small) but never hold a full batch of extracted PDF text in memory at once.
• Idempotency replays. Because Stage III is deterministic, a full recompute of every active registration is cheap and is the recommended nightly reconciliation job rather than incremental in-place mutation.

Failure Modes & Troubleshooting #

Quarantine is a holding state, not a terminal one: once the upstream defect is fixed, the original payload is replayed through process_ingestion and its rejection entry is closed with a forward reference, preserving full lineage.

Compliance Alignment #

This module exists to satisfy concrete statutory obligations, not generic “compliance.” The exposure and filing logic map to named state and federal requirements:

• New York — Executive Law Article 7-A, §172 and §172-b. Charities soliciting in New York register and file the annual CHAR500; organizations with gross revenue and support over $1 million must attach an independent CPA audit. The requires_990 flag and exposure window enforce timely CHAR500 submission within the statutory window.
• California — Government Code §12585 and §12586 (Supervision of Trustees and Fundraisers for Charitable Purposes Act). Initial registration with the Attorney General’s Registry of Charitable Trusts is due within 30 days of receiving charitable assets, and the annual RRF-1 accompanies the IRS Form 990. Under §12586(e)(1), entities with gross annual revenue of $2 million or more must file audited financial statements — a threshold the obligation engine reads from the state matrix.
• IRS Form 990 filing deadline. Renewal windows account for the federal filing deadline of the 15th day of the 5th month after fiscal-year end, since most state registrations require the corresponding 990 as an attachment.
• 2 CFR §200.302 (Financial management). Federal award recipients must maintain records that identify the compliance status of each award; the immutable audit ledger’s terminal_hash provides the tamper-evident trail this section anticipates.

For the full per-jurisdiction matrix of forms, fee tiers, and audit thresholds, follow the State-by-state nonprofit reporting requirements checklist.

Related #

• Core Architecture & Compliance Mapping — parent architecture reference and stage contracts.
• IRS 990 Data Schema Mapping — supplies the canonical, reconciled financial fields this module consumes.
• Grantor-Specific Rule Taxonomies — gates disbursement on the exposure flags emitted here.
• Data Security & Access Boundaries — encryption and IAM enforced at the ingestion edge.
• State-by-state nonprofit reporting requirements checklist — the version-controlled source of truth for forms and thresholds.
• Field Mapping & Normalization — the upstream ingestion stage that feeds this layer.