This guide is part of the State Charity Registration Compliance section within the broader Core Architecture & Compliance Mapping framework, and it solves one narrow problem: how do you stop tracking 50+ jurisdictions of charitable solicitation and annual-reporting obligations in a brittle spreadsheet and start treating the checklist as a deterministic, schema-driven data artifact that a Python pipeline can evaluate, route, and audit on its own?
A reporting-requirements checklist that lives in a shared workbook drifts the moment California changes the RRF-1 fee schedule or New York retunes the CHAR500 audit threshold. Encoded as a version-pinned rule registry instead, the same checklist becomes the single source of truth for threshold evaluation, filing-cadence routing, and immutable audit-trail generation across every state an organization solicits in.
When to Use This Approach
Encode the checklist as a pipeline — rather than maintaining it by hand — when all of the following hold:
- You solicit or report in more than a handful of states. Each jurisdiction carries its own registration threshold, renewal cadence, required form (CA RRF-1, NY CHAR500, IL AG990-IL), and late-filing penalty. Manual tracking becomes unauditable past roughly five states.
- Your filings derive from a stable upstream record. The pipeline assumes inbound data has already cleared the federal ingestion boundary and conforms to the contract sealed by IRS 990 Data Schema Mapping —
ein, total revenue (Form 990 Part VIII line 12), total expenses, and fiscal-year end. Repairing malformed source data is out of scope here. - Obligations are threshold-driven. State registration duties flip on revenue brackets and exemption flags, so the same organization is exempt in one state and owes a full audited return in another. Deterministic evaluation beats human judgment per cycle.
- Every filing must be reconstructable. State Attorneys General audit charitable filings, and IRS Form 990 Schedule O expects disclosure of operational controls. Silent loss of a submission is unacceptable.
If you operate in a single state with a fixed annual obligation, a calendar reminder is cheaper than a pipeline. This approach earns its keep only when the cross-jurisdiction matrix is the source of complexity.
Input preconditions: a raw dict carrying a hyphenated ein, total_revenue, total_expenses, fiscal_year_end, organization_type, and a two-letter jurisdiction_code; a pinned rule_version string; and a writable audit sink. All egress runs over TLS as governed by Data Security & Access Boundaries.
Step-by-Step Implementation
The checklist is enforced in four runnable steps: validate the inbound 990 projection, resolve the state filing directive, dispatch with bounded retry, then seal an immutable audit record. Each stage consumes a typed input and emits a typed output — no mutable state crosses a stage boundary.
Step 1: Validate the 990 projection against state thresholds
State portals change intake APIs and CSV templates without backward compatibility, so the ingestion layer never trusts an unvalidated dict. Inbound payloads are coerced against a rigid Pydantic v2 contract that projects the relevant IRS 990 fields, pins the rule version with a semantic-version pattern, and logs every threshold breach for auditability. The enforce_state_threshold validator carries the per-state registration brackets — CA $250k, NY $100k (audited-return trigger), IL $150k — that the checklist exists to encode.
import json
import logging
from datetime import date
from typing import Any, Dict
from pydantic import BaseModel, Field, ValidationError, field_validator
# Structured audit logger — never print(); compliance events must be parseable.
AUDIT_LOGGER = logging.getLogger("compliance.audit")
AUDIT_LOGGER.setLevel(logging.INFO)
_handler = logging.StreamHandler()
_handler.setFormatter(logging.Formatter("%(asctime)s | %(levelname)s | %(message)s"))
AUDIT_LOGGER.addHandler(_handler)
class IRS990Base(BaseModel):
"""Strictly typed projection of IRS 990 Part I and Part VIII fields."""
ein: str = Field(..., pattern=r"^\d{2}-\d{7}$")
total_revenue: float = Field(..., ge=0.0) # Form 990 Part VIII, line 12
total_expenses: float = Field(..., ge=0.0) # Form 990 Part IX, line 25
fiscal_year_end: date
organization_type: str = Field(..., pattern=r"^(501c3|501c4|501c6)$")
class StateFilingPayload(IRS990Base):
"""990 projection extended with state-specific compliance constraints."""
jurisdiction_code: str = Field(..., min_length=2, max_length=2)
rule_version: str = Field(..., pattern=r"^\d+\.\d+\.\d+$")
solicitation_exempt: bool = False
@field_validator("total_revenue")
@classmethod
def enforce_state_threshold(cls, v: float, info) -> float:
"""Flag organizations above a state's registration/audit threshold.
Thresholds load from a versioned rule registry in production; the
inline map keeps this example deterministic and reproducible.
"""
jurisdiction = info.data.get("jurisdiction_code")
THRESHOLDS: Dict[str, float] = {"CA": 250000.0, "NY": 100000.0, "IL": 150000.0}
threshold = THRESHOLDS.get(jurisdiction, float("inf"))
if v > threshold:
AUDIT_LOGGER.info(json.dumps({
"event": "threshold_exceeded",
"jurisdiction": jurisdiction,
"revenue": v,
"threshold": threshold,
"action": "flag_for_full_registration",
}))
return v
def ingest_and_validate(raw_data: Dict[str, Any], rule_version: str) -> StateFilingPayload:
"""Stage: Ingestion -> Validation. Coerce, pin the rule version, fail loud."""
try:
payload = StateFilingPayload(**raw_data, rule_version=rule_version)
AUDIT_LOGGER.info(json.dumps({
"event": "validation_success", "ein": payload.ein, "rule_version": rule_version,
}))
return payload
except ValidationError as exc:
AUDIT_LOGGER.error(json.dumps({
"event": "validation_failure",
"ein": raw_data.get("ein", "UNKNOWN"),
"errors": exc.errors(),
}))
raise RuntimeError(f"Schema drift detected: {exc}") from exc
The pattern on ein and rule_version is what stops a malformed identifier or an unpinned rule set from silently corrupting every downstream decision. A failed validation raises a typed RuntimeError rather than returning a half-built object.
Step 2: Resolve the state filing directive
A validated payload still has to be mapped to a concrete obligation: which form, which portal, which cadence, and what late-fee schedule. This resolution belongs in the State Charity Registration Compliance layer, and for organizations soliciting in several states it cross-references the per-funder rules captured by Grantor-Specific Rule Taxonomies. Unknown jurisdictions fall through to a MANUAL_REVIEW directive instead of being dropped.
from enum import Enum
from typing import Dict, List
from pydantic import BaseModel
class FilingCadence(str, Enum):
ANNUAL = "annual"
BIENNIAL = "biennial"
EXEMPT = "exempt"
MANUAL_REVIEW = "manual_review"
class JurisdictionDirective(BaseModel):
ein: str
jurisdiction_code: str
cadence: FilingCadence
portal_endpoint: str
required_forms: List[str]
penalty_schedule: Dict[str, float]
def resolve_jurisdiction(payload: StateFilingPayload) -> JurisdictionDirective:
"""Stage: Validation -> Routing. Map a validated payload to a state directive."""
ROUTING_MATRIX: Dict[str, Dict] = {
"CA": {"cadence": FilingCadence.ANNUAL, "portal": "https://oag.ca.gov/charities", "forms": ["RRF-1"]},
"NY": {"cadence": FilingCadence.ANNUAL, "portal": "https://www.charitiesnys.com/", "forms": ["CHAR500"]},
"IL": {"cadence": FilingCadence.ANNUAL, "portal": "https://ilag.gov/", "forms": ["AG990-IL"]},
}
config = ROUTING_MATRIX.get(payload.jurisdiction_code)
if config is None:
AUDIT_LOGGER.warning(json.dumps({
"event": "jurisdiction_unknown", "ein": payload.ein, "state": payload.jurisdiction_code,
}))
return JurisdictionDirective(
ein=payload.ein,
jurisdiction_code=payload.jurisdiction_code,
cadence=FilingCadence.MANUAL_REVIEW,
portal_endpoint="INTERNAL_REVIEW",
required_forms=["MANUAL_INTAKE"],
penalty_schedule={"late_fee": 0.0},
)
# Below the de-minimis solicitation floor, route to EXEMPT instead of filing.
cadence = FilingCadence.EXEMPT if (payload.solicitation_exempt and payload.total_revenue < 25000.0) else config["cadence"]
AUDIT_LOGGER.info(json.dumps({
"event": "routing_resolved", "ein": payload.ein,
"cadence": cadence.value, "forms": config["forms"],
}))
return JurisdictionDirective(
ein=payload.ein,
jurisdiction_code=payload.jurisdiction_code,
cadence=cadence,
portal_endpoint=config["portal"],
required_forms=config["forms"],
penalty_schedule={"late_fee": 100.0, "grace_period_days": 30},
)
Step 3: Dispatch with bounded retry and a circuit breaker
State portals exhibit inconsistent uptime, aggressive rate limiting, and TLS handshake mismatches, so the submission stage separates transient network failure from a permanent compliance rejection. This is the same discipline detailed in Pipeline Fallback & Retry Logic: a 429 or connection reset earns exponential backoff, while a 400/422 schema rejection trips the breaker immediately and escalates to a manual queue.
import time
import requests
from requests.exceptions import RequestException, HTTPError
class CircuitBreakerOpenError(Exception):
pass
class SubmissionEngine:
"""Stage: Routing -> Submission. Retry transient faults, fail fast on rejections."""
def __init__(self, max_retries: int = 3, backoff_factor: float = 1.5) -> None:
self.max_retries = max_retries
self.backoff_factor = backoff_factor
self.circuit_open = False
self.failure_count = 0
self.session = requests.Session()
self.session.headers.update({"User-Agent": "CompliancePipeline/1.0"})
def submit_filing(self, directive: JurisdictionDirective, payload: Dict) -> bool:
if self.circuit_open:
raise CircuitBreakerOpenError("Circuit breaker tripped; manual intervention required.")
for attempt in range(1, self.max_retries + 1):
try:
response = self.session.post(directive.portal_endpoint, json=payload, timeout=30.0)
response.raise_for_status()
AUDIT_LOGGER.info(json.dumps({
"event": "submission_success", "ein": directive.ein,
"attempt": attempt, "status_code": response.status_code,
}))
self.failure_count = 0
return True
except HTTPError as exc:
if response.status_code == 429: # transient: rate limited
wait = self.backoff_factor ** attempt
AUDIT_LOGGER.warning(json.dumps({
"event": "rate_limit_hit", "ein": directive.ein,
"retry_in_seconds": wait, "attempt": attempt,
}))
time.sleep(wait)
continue
if response.status_code in (400, 422): # permanent: schema rejection
AUDIT_LOGGER.error(json.dumps({
"event": "schema_rejection", "ein": directive.ein,
"status_code": response.status_code, "response_body": response.text[:500],
}))
self.failure_count += 1
if self.failure_count >= self.max_retries:
self.circuit_open = True
raise RuntimeError(f"Permanent schema rejection: {exc}") from exc
except RequestException as exc: # transient: network/TLS
wait = self.backoff_factor ** attempt
AUDIT_LOGGER.error(json.dumps({
"event": "network_failure", "ein": directive.ein,
"error": str(exc), "retry_in_seconds": wait,
}))
time.sleep(wait)
self.circuit_open = True
AUDIT_LOGGER.critical(json.dumps({
"event": "submission_exhausted", "ein": directive.ein, "action": "fallback_to_manual_queue",
}))
return False
Step 4: Seal an immutable audit record
Every transaction emits an audit record that a State AG or an external auditor can later verify. The record carries a SHA-256 checksum of the canonical payload, the pinned rule version, the operator context, and a UTC timestamp, conforming to the contract defined by Compliance Metadata Standards. The transaction id is a UUID4 — MD5 and other fast hashes are deliberately avoided for identifiers.
import hashlib
import uuid
from datetime import datetime, timezone
from dataclasses import dataclass
@dataclass(frozen=True)
class ComplianceMetadata:
"""Immutable audit record aligned with state AG audit and IRS 990 retention rules."""
transaction_id: str
ein: str
jurisdiction: str
rule_version: str
payload_checksum: str
timestamp_utc: str
operator_id: str
compliance_status: str
def generate_audit_record(directive: JurisdictionDirective, payload: Dict,
operator_id: str, status: str) -> ComplianceMetadata:
"""Stage: Submission -> Audit. Emit a cryptographically verifiable record."""
payload_bytes = json.dumps(payload, sort_keys=True).encode("utf-8")
checksum = hashlib.sha256(payload_bytes).hexdigest()
tx_id = str(uuid.uuid4())
record = ComplianceMetadata(
transaction_id=tx_id,
ein=directive.ein,
jurisdiction=directive.jurisdiction_code,
rule_version=directive.required_forms[0] if directive.required_forms else "UNKNOWN",
payload_checksum=checksum,
timestamp_utc=datetime.now(timezone.utc).isoformat(),
operator_id=operator_id,
compliance_status=status,
)
AUDIT_LOGGER.info(json.dumps({
"event": "audit_record_generated", "transaction_id": tx_id,
"checksum": checksum, "status": status,
}))
return record
Verification
Confirm the checklist behaves deterministically before trusting it in a filing cycle:
- Threshold flag fires. Feed a California payload with
total_revenueof300000.0; the audit log must contain athreshold_exceededevent with"threshold": 250000.0and"action": "flag_for_full_registration". - Exemption routing holds. A payload with
solicitation_exempt=Trueandtotal_revenuebelow25000.0must resolve toFilingCadence.EXEMPT, notANNUAL. - Unknown states never drop. A
jurisdiction_codeoutside the routing matrix (e.g."WY") must produce aMANUAL_REVIEWdirective and ajurisdiction_unknownwarning — never a silent pass. - Checksums are stable. Calling
generate_audit_recordtwice on an identical payload must yield an identicalpayload_checksum(thesort_keys=Truecanonicalization guarantees it) while thetransaction_iddiffers each time. - Idempotent across environments. Run the same fixture in staging and production with the same
rule_version; routing decisions and checksums must match byte-for-byte. Divergence means a rule registry is unpinned.
A minimal pytest assertion for the first check:
def test_ca_threshold_flag(caplog):
payload = ingest_and_validate(
{"ein": "12-3456789", "total_revenue": 300000.0, "total_expenses": 280000.0,
"fiscal_year_end": "2025-12-31", "organization_type": "501c3", "jurisdiction_code": "CA"},
rule_version="2.4.0",
)
assert payload.total_revenue == 300000.0
assert any("threshold_exceeded" in r.message for r in caplog.records)
Common Errors & Fixes
| Error | Cause | Fix |
|---|---|---|
ValidationError: ein string does not match pattern |
EIN passed without the hyphen (123456789) |
Normalize to NN-NNNNNNN at ingestion; reject, never auto-strip silently |
threshold_exceeded never logged for a high-revenue org |
jurisdiction_code missing from the THRESHOLDS map, so threshold defaults to inf |
Add the state to the versioned rule registry; treat unmapped states as MANUAL_REVIEW, not exempt |
| Org filed in the wrong state | Routing keyed off mailing address instead of solicitation activity | Resolve jurisdiction_code from where the org solicits, per each state’s charitable-solicitation statute |
CircuitBreakerOpenError on every submission |
A prior 400/422 tripped the breaker and it was never reset |
Resolve the schema rejection, then re-instantiate SubmissionEngine (or add an explicit half-open reset) |
| Audit checksum differs across runs of identical data | Payload serialized without sort_keys=True, so key order varies |
Always canonicalize with json.dumps(payload, sort_keys=True) before hashing |
| Filing accepted but flagged late by the state | rule_version pinned to a stale fee/deadline schedule |
Tag every rule registry with a semantic version and fail CI on divergence from the current cycle |
For authoritative federal guidance, reference the IRS Form 990 Instructions; for Python logging discipline in compliance pipelines, the standard-library logging module; and for multi-state coordination, the National Association of Attorneys General.
Related
- State Charity Registration Compliance — the parent section this checklist plugs into.
- How to Map IRS 990 Part VII to JSON Schema — the upstream contract that feeds the validated payload used here.
- Building a Fallback Routing System for Grant APIs — the retry and circuit-breaker pattern this guide reuses for state-portal submission.
- Standardizing Grant Field Names Across Multiple Portals — cross-pillar field-alias resolution for inconsistent state intake formats.