Grantor-Specific Rule Taxonomies: Stage-Specific Implementation Guide

Positioned as a discrete operational subsystem within the broader Core Architecture & Compliance Mapping framework, the Grantor-Specific Rule Taxonomy engine operates exclusively at the rule evaluation stage. This guide isolates taxonomy compilation, validation, and execution from upstream ingestion, financial reconciliation, and downstream reporting. The objective is to establish deterministic, auditable rule evaluation for grantor-specific compliance requirements without cross-contaminating pipeline boundaries. Nonprofit operations teams, grant managers, Python automation developers, and compliance officers must treat this stage as a stateless, deterministic evaluation layer that consumes normalized payloads and emits structured compliance verdicts.

Architectural Positioning & Stage Boundaries #

The rule taxonomy stage accepts only reconciled, schema-validated payloads. It does not perform raw data parsing, currency conversion, or final regulatory report generation. All inputs must conform to pre-validated canonical schemas. Outputs are strictly compliance verdicts, exception flags, and structured audit metadata. Blending ingestion, reconciliation, rule evaluation, or reporting logic into a single execution context violates pipeline isolation principles and introduces non-deterministic state drift. Each stage must maintain explicit interface contracts, versioned schemas, and independent error boundaries.

Boundary enforcement is non-negotiable:

• Upstream Contract: Payloads arrive pre-normalized, with all monetary values converted to base currency units and dates standardized to UTC.
• Internal Contract: The taxonomy engine evaluates conditions against a frozen execution context. No external I/O, network calls, or mutable global state is permitted during evaluation.
• Downstream Contract: The engine emits a deterministic verdict payload containing pass/fail status, triggered rule identifiers, exception severity, and cryptographic audit signatures.

Procedural Workflow #

1. Taxonomy Ingestion & Explicit Schema Validation #

Grantor rule definitions must be loaded from a version-controlled registry (JSON or YAML) and validated before compilation. Explicit validation gates enforce structural integrity and prevent malformed taxonomies from entering the execution graph.

• Schema Enforcement: Utilize pydantic or jsonschema to validate required fields: rule_id, grantor_code, condition_expression, compliance_threshold, effective_date, expiration_date, and priority.
• Type Coercion & Rejection: Reject payloads with ambiguous types immediately. Floating-point thresholds must be coerced to decimal.Decimal to prevent IEEE 754 rounding artifacts. Date fields must parse strictly to ISO 8601 UTC.
• Deterministic Error Codes: Return explicit failure states (ERR_TAXONOMY_SCHEMA_INVALID, ERR_GRANTOR_CODE_MISMATCH, ERR_DATE_RANGE_OVERLAP) without attempting heuristic recovery.
• Canonical Tooling: pydantic, jsonschema, ruamel.yaml, datetime, decimal.

2. Rule Compilation & Deterministic Execution #

Once validated, taxonomies are compiled into an ordered execution graph. Rule evaluation must remain strictly deterministic, reproducible, and isolated from external state mutations.

• Safe Expression Parsing: Restrict condition evaluation to a sandboxed AST parser. Prohibit arbitrary Python execution, network calls, or filesystem access within rule expressions.
• Execution Ordering: Sort rules deterministically by priority (descending), effective_date (ascending), and rule_id (lexicographic). This guarantees identical evaluation sequences across distributed workers.
• Context Freezing: Bind all evaluation variables to an immutable mapping. Any attempt to mutate context during evaluation raises a RuntimeError and aborts the batch.

3. Verdict Emission & Structured Audit Hooks #

Evaluation results are aggregated into a standardized verdict envelope. Every rule evaluation generates an immutable audit record containing input snapshots, evaluation timestamps, and compliance flags.

• Verdict Structure: { "evaluation_id": "uuid", "grantor_code": "str", "verdict": "PASS|FAIL|PARTIAL", "triggered_rules": ["rule_id"], "compliance_score": "decimal", "audit_trail": [...] }
• Audit Hooks: Attach cryptographic hashes of the input payload and taxonomy version to each verdict. This enables downstream compliance officers to reconstruct evaluation states without re-executing the pipeline.

Production Implementation Reference #

The following module demonstrates a production-grade implementation adhering to strict separation of concerns, deterministic execution, and embedded audit hooks. It relies on pydantic for schema validation, decimal for financial precision, and Python’s ast module for sandboxed expression evaluation.

import ast
import uuid
import logging
import datetime
from decimal import Decimal, InvalidOperation
from typing import Any, Dict, List, Optional
from pydantic import BaseModel, Field, field_validator

logger = logging.getLogger(__name__)

# --- Schema Definitions ---
class GrantorRule(BaseModel):
    rule_id: str = Field(pattern=r"^[A-Z0-9_]{4,32}$")
    grantor_code: str = Field(min_length=3, max_length=12)
    condition_expression: str
    compliance_threshold: Decimal
    effective_date: datetime.datetime
    expiration_date: datetime.datetime
    priority: int = Field(ge=1, le=100)

    @field_validator("effective_date", "expiration_date", mode="before")
    @classmethod
    def enforce_utc(cls, v: Any) -> datetime.datetime:
        if isinstance(v, str):
            dt = datetime.datetime.fromisoformat(v.replace("Z", "+00:00"))
            if dt.tzinfo is None or dt.utcoffset() != datetime.timedelta(0):
                raise ValueError("Dates must be ISO 8601 UTC")
            return dt
        return v

    @field_validator("compliance_threshold", mode="before")
    @classmethod
    def coerce_decimal(cls, v: Any) -> Decimal:
        try:
            return Decimal(str(v))
        except InvalidOperation:
            raise ValueError("Threshold must be a valid numeric string or Decimal")

# --- Safe Evaluator ---
class SandboxedEvaluator:
    ALLOWED_NODES = {
        ast.Expression, ast.BinOp, ast.UnaryOp, ast.Compare,
        ast.Num, ast.Constant, ast.Name, ast.Load,
        ast.Add, ast.Sub, ast.Mult, ast.Div,
        ast.Lt, ast.LtE, ast.Gt, ast.GtE, ast.Eq, ast.NotEq
    }

    @classmethod
    def validate_ast(cls, source: str) -> ast.Expression:
        tree = ast.parse(source, mode="eval")
        for node in ast.walk(tree):
            if type(node) not in cls.ALLOWED_NODES:
                raise SyntaxError(f"Unsafe AST node detected: {type(node).__name__}")
        return tree

    @classmethod
    def evaluate(cls, expression: str, context: Dict[str, Decimal]) -> bool:
        tree = cls.validate_ast(expression)
        # Restrict builtins to prevent accidental function calls
        safe_globals = {"__builtins__": {}}
        return bool(eval(compile(tree, "<rule_expr>", "eval"), safe_globals, context))

# --- Audit Hook ---
class AuditRecord(BaseModel):
    rule_id: str
    expression: str
    verdict: bool
    evaluated_at: datetime.datetime
    context_hash: str  # SHA-256 of frozen input context

# --- Execution Engine ---
class TaxonomyEngine:
    def __init__(self, rules: List[GrantorRule]):
        self._rules = sorted(rules, key=lambda r: (-r.priority, r.effective_date, r.rule_id))
        self._compiled = []
        self._compile()

    def _compile(self):
        for rule in self._rules:
            try:
                tree = SandboxedEvaluator.validate_ast(rule.condition_expression)
                self._compiled.append((rule, tree))
            except SyntaxError as e:
                logger.error(f"ERR_TAXONOMY_SCHEMA_INVALID: {rule.rule_id} - {e}")
                raise

    def evaluate(self, payload: Dict[str, Decimal], evaluation_id: Optional[str] = None) -> Dict[str, Any]:
        eval_id = evaluation_id or str(uuid.uuid4())
        triggered = []
        audit_trail = []
        context_hash = str(hash(frozenset(payload.items())))  # Deterministic snapshot

        for rule, tree in self._compiled:
            context = {k: v for k, v in payload.items()}
            try:
                result = SandboxedEvaluator.evaluate(rule.condition_expression, context)
                if result:
                    triggered.append(rule.rule_id)
                audit_trail.append(AuditRecord(
                    rule_id=rule.rule_id,
                    expression=rule.condition_expression,
                    verdict=result,
                    evaluated_at=datetime.datetime.now(datetime.timezone.utc),
                    context_hash=context_hash
                ).model_dump())
            except Exception as e:
                logger.error(f"ERR_EVALUATION_FAILURE: {rule.rule_id} -> {e}")
                raise

        return {
            "evaluation_id": eval_id,
            "grantor_code": self._rules[0].grantor_code if self._rules else "UNKNOWN",
            "verdict": "PASS" if not triggered else "FAIL",
            "triggered_rules": triggered,
            "audit_trail": audit_trail,
            "taxonomy_version": "v1.0.0"
        }

Compliance Mapping & Pipeline Handoff Protocols #

The taxonomy engine does not generate regulatory filings. It produces structured verdicts that downstream compliance systems consume. Handoff boundaries are explicitly defined to maintain audit integrity and regulatory alignment.

1. Federal Reporting Handoff: When a verdict indicates FAIL or PARTIAL, the payload is routed to the IRS 990 Data Schema Mapping pipeline. The audit trail provides the exact rule expressions and threshold comparisons required for Part IX functional expense allocation and Schedule O narrative justification.
2. State-Level Registration Handoff: Verdicts containing grantor_code values mapped to multi-state jurisdictions trigger conditional routing to the State Charity Registration Compliance subsystem. The deterministic execution ordering ensures that conflicting state-level thresholds are resolved before jurisdictional filing generation.
3. Security & Access Boundaries: Audit records are serialized with immutable context hashes. Read access to the evaluation graph is restricted to compliance officers and authorized automation service accounts. Write access to the taxonomy registry requires cryptographic signing and version control merge gates.
4. Fallback & Retry Logic: If the evaluation engine encounters a transient dependency failure (e.g., registry fetch timeout), the pipeline invokes Pipeline Fallback & Retry Logic to queue the payload. The engine itself remains stateless; retries re-evaluate against the same frozen context, guaranteeing idempotent verdicts.

Operational Guardrails & Edge Case Handling #

• Threshold Precision: All monetary comparisons use decimal.Decimal with explicit quantization. Never use float for compliance thresholds. Refer to Python’s decimal documentation for context manager configuration (decimal.getcontext().prec = 28).
• Expression Sandboxing: The AST validator explicitly blocks attribute access (ast.Attribute), function calls (ast.Call), and imports (ast.Import). This prevents arbitrary code execution and aligns with Pydantic’s strict type validation principles.
• Date Boundary Overlaps: If effective_date and expiration_date create overlapping windows for identical grantor_code and priority values, the ingestion gate raises ERR_DATE_RANGE_OVERLAP. Overlaps introduce non-deterministic rule precedence and must be resolved at the registry level.
• Audit Retention: Verdict payloads must be retained for a minimum of seven fiscal years. The audit_trail array contains all necessary metadata to reconstruct evaluation states without storing raw grantor payloads in the compliance archive.
• Deterministic Error Codes: All failure states map to machine-readable codes. Heuristic recovery is prohibited. If a rule fails validation, the entire taxonomy batch is rejected, logged, and routed to a quarantine queue for manual compliance review.

By enforcing strict stage boundaries, deterministic execution ordering, and cryptographically verifiable audit trails, the Grantor-Specific Rule Taxonomy engine delivers reliable, compliance-ready evaluation outputs. Nonprofit operations teams can integrate this subsystem with confidence, knowing that pipeline handoffs preserve data integrity and regulatory alignment across federal and state reporting requirements.