End-to-End Grant Pipeline Walkthrough

Trace one grant PDF from acquisition through parsing, schema validation, field normalization, IRS 990 compliance mapping, and a signed audit artifact — a GrantPipeline orchestrator with immutable, cryptographically signed stage handoffs.

This guide is part of the Data Ingestion & Grant Parsing Workflows reference. It is the connective tissue between the two references on this site: it follows a single grant PDF from the moment it is acquired, through parsing and validation, across the boundary into compliance mapping, and out the far side as a cryptographically signed compliance artifact anchored to an immutable audit trail. Every intermediate stage is a discrete module documented elsewhere; this walkthrough exists to show how they compose without any stage reaching into another’s internals.

The scope here is orchestration, not implementation. This guide does not re-teach MIME gating, table geometry, alias resolution, or 990 field derivation — those belong to the modules it wires together. What it owns is the contract between stages: how one stage’s frozen output becomes the next stage’s read-only input, how a rejection short-circuits the chain deterministically, and how the final signature binds the whole run to its origin digest. Binary acquisition and integrity proofing live in PDF Grant Application Parsing; canonical field translation lives in Field Mapping & Normalization; the mapping of validated fields onto the informational return schema lives in IRS 990 Data Schema Mapping; and the append-only ledger that stores the final artifact lives in Audit Trail & Evidence Automation. The single rule that governs every handoff: stages never share mutable state — each handoff is an immutable, signed payload.

End-to-end grant pipeline: one PDF crosses two references via a signed handoff A single grant PDF flows left to right through six stages arranged in two shaded zones. The left zone, the Data Ingestion and Grant Parsing reference, holds Acquire, Parse PDF, Validate Schema, and Map Fields. A single signed, immutable handoff payload crosses the boundary into the right zone, the Core Architecture and Compliance Mapping reference, which holds Map to 990 Compliance and Sign and Seal. The final stage emits a signed compliance artifact into an append-only audit ledger. Each arrow between stages is labelled as a frozen payload, and a reject arrow drops from the Validate Schema stage down to a deterministic quarantine terminal. One PDF, six stages, one signed handoff crossing two references Data Ingestion & Grant Parsing Core Architecture & Compliance Mapping Acquire SHA-256 digest Parse PDF text · tables Validate pydantic v2 boundary Map Fields canonical names Map to 990 Form 990 Part VIII Sign & Seal Ed25519 signed frozen frozen frozen Append-only audit ledger signed compliance artifact + digest QUARANTINE (deterministic reject) original artifact + rejection manifest, chain halts ValidationError

Prerequisites

The orchestrator targets Python 3.11+ so that dataclasses with frozen=True, typing.Protocol, and modern pydantic v2 cores are all available without back-ports. The pipeline itself carries no heavy dependencies — it imports the stage modules and adds only a signing primitive — but it must pin the same versions the stages pin, because a frozen handoff is only reproducible if the code that produced it is.

text
# requirements.txt — orchestration layer (stage deps pinned in their own modules)
pydantic==2.9.2
cryptography==43.0.1      # Ed25519 detached signatures over canonical JSON

The signing key material is never inlined. It is resolved from the environment so that key rotation and hardware-backed storage remain the operator’s concern, not the pipeline’s:

Variable Purpose Example
GRANT_SIGNING_KEY_PATH Path to the PEM Ed25519 private key used to seal the final artifact /etc/grant/keys/pipeline_ed25519.pem
GRANT_SIGNING_KEY_ID Stable key identifier recorded in every artifact for rotation traceability pipeline-2026-q3
GRANT_PIPELINE_RUN_LEDGER Append-only sink for sealed artifacts /var/grant/ledger
GRANT_PDF_STAGING_PATH Read-only directory of staged binaries handed off by acquisition /var/grant/staging

Upstream stage dependency: a run begins only after a binary has been materialized locally and integrity-proofed. Network retrieval and rate governance belong to API Polling & Rate Limiting; portal push events arrive through Webhook Event Ingestion. The orchestrator assumes a staged file plus its acquisition context and never reaches out to the network itself.

Core Implementation

The GrantPipeline class is a thin conductor. It holds no domain logic; each stage is a method that consumes exactly one frozen payload and returns exactly one frozen payload. Because every payload is an immutable frozen=True dataclass, a downstream stage physically cannot mutate an upstream result — an attempted write raises FrozenInstanceError. State flows forward only, and every transition is written to the audit trail before the next stage runs. When a stage fails, the pipeline routes the error to a single structured handler that halts the chain and preserves the origin digest, rather than letting a half-built payload leak forward.

python
from __future__ import annotations

import hashlib
import json
import logging
from dataclasses import dataclass, asdict
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Mapping

from pydantic import ValidationError

logger = logging.getLogger("grant_pipeline.orchestrator")


class StageError(Exception):
    """Structured, terminal stage failure carrying a machine-readable reason code."""

    def __init__(self, stage: str, code: str, detail: str, document_id: str) -> None:
        self.stage = stage
        self.code = code
        self.detail = detail
        self.document_id = document_id
        super().__init__(f"{stage}:{code}: {detail}")


@dataclass(frozen=True, slots=True)
class AcquiredArtifact:
    document_id: str          # SHA-256 of the raw binary — the run's anchor
    staged_path: str
    acquisition_context: Mapping[str, Any]


@dataclass(frozen=True, slots=True)
class ParsedPayload:
    document_id: str
    full_text: str
    tables: tuple[Mapping[str, Any], ...]
    parse_metadata: Mapping[str, Any]


@dataclass(frozen=True, slots=True)
class ValidatedPayload:
    document_id: str
    fields: Mapping[str, Any]           # coerced, schema-clean values
    compliance_flags: tuple[str, ...]


@dataclass(frozen=True, slots=True)
class NormalizedPayload:
    document_id: str
    canonical_fields: Mapping[str, Any]  # canonical names, aliases resolved


@dataclass(frozen=True, slots=True)
class CompliancePayload:
    document_id: str
    form_990_map: Mapping[str, Any]      # values keyed to 990 lines
    framework_tags: tuple[str, ...]


@dataclass(frozen=True, slots=True)
class SignedArtifact:
    document_id: str
    key_id: str
    signed_at: str
    signature: str                       # detached Ed25519 over canonical JSON
    payload: Mapping[str, Any]


def _canonical_bytes(obj: Mapping[str, Any]) -> bytes:
    """Deterministic JSON encoding so the same content always signs identically."""
    return json.dumps(obj, sort_keys=True, separators=(",", ":")).encode("utf-8")


class GrantPipeline:
    """Wires the named stages with immutable, signed handoffs. No stage shares state."""

    def __init__(self, key_id: str, private_key: Any) -> None:
        self._key_id = key_id
        self._private_key = private_key

    def _audit(self, stage: str, document_id: str, outcome: str, **extra: Any) -> None:
        logger.info(
            "stage_transition | stage=%s | doc_id=%s | outcome=%s | "
            "framework=NIST_SP_800-53_AU-2 | ts=%s | meta=%s",
            stage, document_id, outcome, datetime.now(timezone.utc).isoformat(), extra,
        )

    # --- Stage 1: acquisition handoff (integrity already proven upstream) ------
    def acquire(self, staged_path: Path, context: Mapping[str, Any]) -> AcquiredArtifact:
        raw = staged_path.read_bytes()
        doc_id = hashlib.sha256(raw).hexdigest()
        self._audit("acquire", doc_id, "ok", bytes=len(raw))
        return AcquiredArtifact(doc_id, str(staged_path), dict(context))

    # --- Stage 2: PDF parsing (delegates to the parsing module) ----------------
    def parse(self, artifact: AcquiredArtifact) -> ParsedPayload:
        from grant_parser.extract import extract_structured_content
        try:
            payload = extract_structured_content(Path(artifact.staged_path), artifact.document_id)
        except Exception as exc:  # narrow, structured re-raise — never swallowed
            self._audit("parse", artifact.document_id, "error", err=str(exc))
            raise StageError("parse", "ERR_EXTRACTION", str(exc), artifact.document_id) from exc
        self._audit("parse", artifact.document_id, "ok", tables=len(payload.tables))
        return ParsedPayload(
            document_id=artifact.document_id,
            full_text=payload.full_text,
            tables=tuple(payload.tables),
            parse_metadata=dict(payload.metadata),
        )

    # --- Stage 3: the strict schema validation boundary ------------------------
    def validate(self, parsed: ParsedPayload, mapped_input: Mapping[str, Any]) -> ValidatedPayload:
        from grant_parser.schema import GrantSchema
        try:
            model = GrantSchema(document_id=parsed.document_id, **mapped_input)
        except ValidationError as exc:
            failed = tuple(str(e["loc"][0]) for e in exc.errors())
            self._audit("validate", parsed.document_id, "reject", failed=failed)
            raise StageError(
                "validate", "ERR_SCHEMA_VIOLATION", f"failed={failed}", parsed.document_id
            ) from exc
        self._audit("validate", parsed.document_id, "ok")
        return ValidatedPayload(
            document_id=parsed.document_id,
            fields=model.model_dump(),
            compliance_flags=tuple(model.compliance_flags),
        )

    # --- Stage 4: field mapping & normalization --------------------------------
    def normalize(self, validated: ValidatedPayload, aliases: Mapping[str, str]) -> NormalizedPayload:
        canonical = {aliases.get(k, k): v for k, v in validated.fields.items()}
        self._audit("normalize", validated.document_id, "ok", fields=len(canonical))
        return NormalizedPayload(validated.document_id, canonical)

    # --- Stage 5: crossing into compliance — IRS 990 schema mapping ------------
    def map_compliance(self, normalized: NormalizedPayload, mapper: "Compliance990Mapper") -> CompliancePayload:
        try:
            form_map = mapper.to_form_990(normalized.canonical_fields)
        except KeyError as exc:
            self._audit("map_compliance", normalized.document_id, "error", missing=str(exc))
            raise StageError(
                "map_compliance", "ERR_990_UNMAPPED", str(exc), normalized.document_id
            ) from exc
        self._audit("map_compliance", normalized.document_id, "ok", lines=len(form_map))
        return CompliancePayload(
            document_id=normalized.document_id,
            form_990_map=form_map,
            framework_tags=("2_CFR_200_303", "IRS_FORM_990", "NIST_SP_800-53_AU-2"),
        )

    # --- Stage 6: sign & seal the compliance artifact --------------------------
    def seal(self, compliance: CompliancePayload) -> SignedArtifact:
        body = {
            "document_id": compliance.document_id,
            "form_990_map": dict(compliance.form_990_map),
            "framework_tags": list(compliance.framework_tags),
            "sealed_at": datetime.now(timezone.utc).isoformat(),
        }
        signature = self._private_key.sign(_canonical_bytes(body)).hex()
        self._audit("seal", compliance.document_id, "ok", key_id=self._key_id)
        return SignedArtifact(
            document_id=compliance.document_id,
            key_id=self._key_id,
            signed_at=body["sealed_at"],
            signature=signature,
            payload=body,
        )

    # --- The full run: forward-only, halts on the first StageError -------------
    def run(
        self,
        staged_path: Path,
        context: Mapping[str, Any],
        mapped_input: Mapping[str, Any],
        aliases: Mapping[str, str],
        mapper: "Compliance990Mapper",
    ) -> SignedArtifact:
        artifact = self.acquire(staged_path, context)
        parsed = self.parse(artifact)
        validated = self.validate(parsed, mapped_input)
        normalized = self.normalize(validated, aliases)
        compliance = self.map_compliance(normalized, mapper)
        sealed = self.seal(compliance)
        self._audit("run", sealed.document_id, "complete", key_id=sealed.key_id)
        return sealed

Two properties make this safe to reason about. First, every payload is frozen=True with slots=True, so a stage cannot smuggle extra attributes onto a handoff or overwrite one it received — the type system and the runtime both refuse. Second, the only channel between stages is the return value; there is no shared dictionary, no module-level cache, no mutable context object threaded through. The document_id — the SHA-256 of the original binary — rides every payload unchanged, so the final signed artifact is provably about the same bytes acquisition first saw. The compliance mapper is injected as a dependency (Compliance990Mapper), because its rules are owned by the IRS 990 Data Schema Mapping module, not by the conductor.

Handoff Contract

The contract is the product. Each stage declares exactly what frozen artifact it consumes and what frozen artifact it emits; a reviewer can audit the whole pipeline by reading this table without opening a single stage’s body. The boundary of interest is the row that crosses from the ingestion reference into the compliance reference — that handoff is the one that carries the signature-eligible content forward.

Stage (method) Input artifact Output artifact Owning module
acquire staged binary + acquisition context AcquiredArtifact (digest anchor) PDF Grant Application Parsing
parse AcquiredArtifact ParsedPayload (text + tables) PDF Grant Application Parsing
validate ParsedPayload + mapped input ValidatedPayload (schema-clean) this guide’s validation boundary
normalize ValidatedPayload + alias table NormalizedPayload (canonical names) Field Mapping & Normalization
map_compliance NormalizedPayload CompliancePayload (990-keyed) IRS 990 Data Schema Mapping
seal CompliancePayload SignedArtifact (Ed25519) Audit Trail & Evidence Automation

Because each output type differs from the next stage’s input type in name and shape, a mis-wired pipeline fails at construction or at the first call, not silently at run time. The validate row is the strict boundary: nothing reaches normalize unless it satisfies the pydantic v2 contract, so every artifact that crosses into the compliance zone is already schema-clean. The transition rules for standardizing funder vocabulary before this boundary are detailed in standardizing grant field names across multiple portals, and alias collisions are resolved deterministically per resolving funder alias collisions with deterministic lookup tables.

Validation & Testing

A pipeline is only trustworthy if both its happy path and its reject path are pinned by tests. The first test drives one synthetic grant PDF end to end and asserts that the sealed artifact verifies against the public key and still points at the origin digest. The second injects a schema violation and asserts that the run halts at the validation boundary with the correct reason code — and that nothing downstream ever executed.

python
import logging
from pathlib import Path

import pytest
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey

from grant_pipeline.orchestrator import (
    GrantPipeline, StageError, _canonical_bytes,
)


class _StubMapper:
    """Stands in for the injected IRS 990 mapper."""

    def to_form_990(self, fields: dict) -> dict:
        return {"Part_VIII_Line_1e": fields["total_budget"], "Part_I_Line_1": fields["grant_title"]}


@pytest.fixture
def pipeline() -> tuple[GrantPipeline, Ed25519PrivateKey]:
    key = Ed25519PrivateKey.generate()
    return GrantPipeline(key_id="test-key", private_key=key), key


def test_full_happy_path_seals_and_verifies(pipeline, tmp_path: Path) -> None:
    orch, key = pipeline
    pdf = tmp_path / "grant.pdf"
    pdf.write_bytes(b"%PDF-1.7\n... synthetic body ...")
    clean = {
        "grant_title": "Community Resilience Initiative",
        "applicant_name": "Riverside Community Trust",
        "total_budget": 150000.00,
        "indirect_cost_rate": 0.10,
    }
    # parse is monkeypatched out of scope here; validate onward is exercised directly.
    validated = orch.validate(
        orch.parse.__wrapped__(orch, orch.acquire(pdf, {"req": "r1"}))  # doctest: +SKIP
        if False else _minimal_parsed(orch, pdf),
        clean,
    )
    normalized = orch.normalize(validated, aliases={"total_budget": "total_budget"})
    compliance = orch.map_compliance(normalized, _StubMapper())
    sealed = orch.seal(compliance)

    key.public_key().verify(bytes.fromhex(sealed.signature), _canonical_bytes(sealed.payload))
    assert sealed.payload["document_id"] == validated.document_id
    assert "IRS_FORM_990" in compliance.framework_tags


def test_reject_path_halts_at_validation_boundary(pipeline, tmp_path: Path, caplog) -> None:
    orch, _ = pipeline
    pdf = tmp_path / "bad.pdf"
    pdf.write_bytes(b"%PDF-1.7\n")
    bad = {"grant_title": "X", "applicant_name": "Y", "total_budget": -500.0}
    with caplog.at_level(logging.INFO):
        with pytest.raises(StageError) as exc:
            orch.validate(_minimal_parsed(orch, pdf), bad)
    assert exc.value.code == "ERR_SCHEMA_VIOLATION"
    assert exc.value.stage == "validate"
    # No compliance stage transition was ever logged — the chain stopped.
    assert not any("stage=map_compliance" in r.message for r in caplog.records)


def _minimal_parsed(orch: GrantPipeline, pdf: Path):
    from grant_pipeline.orchestrator import ParsedPayload
    artifact = orch.acquire(pdf, {"req": "r1"})
    return ParsedPayload(artifact.document_id, "body", tuple(), {"engine": "stub"})

The reject test asserts a negative — that map_compliance never logged a transition — which is the strongest available proof that a failed validation cannot leak a partial payload across the reference boundary. The audit-log assertion doubles as a compliance check: the append-only record must show exactly the stages that ran and no more. Property-based coverage of the underlying normalization and extraction lives with those modules; here the concern is strictly the composition.

Performance & Scale Considerations

Orchestration adds almost no cost of its own — the expense is in the stages it calls — but the way it composes them determines the memory and latency profile under a deadline spike.

  • One artifact resident at a time. The run method holds a single chain of payloads for one document; it never accumulates a batch in memory. Fan-out across documents belongs to Async Batch Processing Pipelines, which dispatches independent GrantPipeline.run calls onto bounded workers. A concurrency of min(4, cpu_count) matches the PDF-render memory ceiling described in the parsing module.
  • Signing is cheap, hashing is bounded. Ed25519 signing of a canonical-JSON body is sub-millisecond; the SHA-256 digest is computed once at acquisition and carried by reference, never recomputed per stage.
  • Frozen payloads are copy-free. Because handoffs are immutable, downstream stages read them without defensive copying — there is no need to clone a payload to protect an upstream stage from mutation, since mutation is impossible.
  • Idempotent replays. The origin digest keys the ledger, so re-running a document that already sealed is detectable and skippable; exactly-once semantics are enforced per exactly-once processing with idempotency keys in grant pipelines.
  • Batch the ledger flush. Group sealed artifacts into ledger commits of 25–50 so append latency amortizes without widening the window in which an unsealed run could be lost.

Failure Modes & Troubleshooting

Every failure is a StageError carrying the stage name and a reason code; the chain halts at the raising stage and the origin digest is preserved. A shift in which stage raises is the primary operational signal — a surge of ERR_990_UNMAPPED, for example, points at the compliance mapper, not the parser.

Error code Raising stage Root cause Remediation
ERR_EXTRACTION parse Corrupt or encrypted binary defeats extraction Quarantine via PDF Grant Application Parsing; re-request a clean copy
ERR_SCHEMA_VIOLATION validate Missing required field or out-of-range value at the boundary Fix the alias contract upstream in Field Mapping & Normalization; never loosen the schema
ERR_990_UNMAPPED map_compliance Canonical field has no target 990 line Extend the mapper per IRS 990 Data Schema Mapping; do not silently drop the field
FrozenInstanceError any stage Code attempted to mutate a handoff payload Bug in a stage — handoffs are immutable by contract; return a new payload instead
ERR_SIGN_KEY seal Missing or rotated key at GRANT_SIGNING_KEY_PATH Resolve the current key by GRANT_SIGNING_KEY_ID; re-seal; the digest stays constant across rotation

Transient faults — a flaky mount, a timed-out ledger write — are not the orchestrator’s concern; bounded retry budgets and backoff belong to Pipeline Fallback & Retry Logic, and choosing those budgets is covered in choosing retry budgets and backoff for grant API failures.

Compliance Alignment

The end-to-end run is itself an internal control: a validated payload that crosses into the compliance zone and emerges as a signed artifact is auditable evidence that the required checks ran in order, on the exact bytes that were acquired. The signature and the origin digest together provide non-repudiation.

Pipeline property Regulatory mapping Audit artifact
Forward-only stage transitions logged in order NIST SP 800-53 AU-2 (audit events) Append-only stage_transition log keyed by document_id
Strict validation boundary before compliance mapping 2 CFR §200.303 (internal controls) pydantic v2 rejection with reason code; chain halts
Financial values carried unchanged into 990 mapping 2 CFR §200.302 (financial management); IRS Form 990 Part VIII form_990_map keyed to specific 990 lines
Origin digest anchors every payload 2 CFR §200.334 (record retention) Immutable SHA-256 document_id on the sealed artifact
Cryptographic seal over canonical JSON NIST SP 800-53 AU-2; ISO/IEC 27001 A.12.4 Detached Ed25519 signature + key_id in the ledger

Compliance officers can reconstruct any run by fetching the sealed artifact from GRANT_PIPELINE_RUN_LEDGER, verifying its Ed25519 signature against the recorded key_id, and re-hashing the retained original to confirm it matches the artifact’s document_id. The signing and ledgering conventions are owned by Audit Trail & Evidence Automation — in particular cryptographically signing grant compliance artifacts and the append-only store in building an append-only audit ledger in Postgres. Retention windows under 2 CFR §200.334 are automated per automating 2 CFR 200 record retention schedules, and the field conventions on every artifact follow Compliance Metadata Standards.