Seasonal Compliance Report: From Telemetry to EPA Submission

Problem statement

At the end of a spray season an operations team has to hand a regulator a defensible account of every restricted-use application: what was applied, at what rate, on which field, when, and under whose authority. The raw material for that account is machine telemetry — thousands of ISOBUS and OEM API records that were never written with a filing in mind. The gap between “the sprayers reported this” and “here is a filed EPA package” is a multi-stage pipeline, and every stage is a place where a plausible-but-wrong number can slip through and only surface during the inspection it was meant to satisfy.

This guide walks the whole path as one orchestrated workflow: raw telemetry is normalized into canonical records, each record is validated against the active rule set, valid records are written to the tamper-evident audit ledger, a jurisdiction-shaped PDF and CSV are rendered from the ledger slice, and the two are assembled into a submission package with a manifest hash. The upstream normalization of machine payloads is owned by Equipment Telemetry Parsing, and the rules each record is checked against come from EPA/USDA Rule Mapping; this page is the orchestration that stitches those upstream contracts to a filed artifact. It assumes the ledger and export mechanics described in the parent Audit & Reporting topic.

Parameter reference

Each value below changes what lands in the filed package. Recommended values assume a single-season EPA restricted-use filing and an intermittently connected fleet.

Runnable orchestration

The module below orchestrates the six stages as one typed pipeline. It targets Python 3.10+, is fully typed, and depends on pydantic>=2.5 for the record model and the ledger and renderer described in the parent topic. The upstream normalize_telemetry and render_package functions are shown by signature — their internals belong to the parsing and export guides — so this file is the coordination layer, not a re-implementation.

from __future__ import annotations

import hashlib
import json
import logging
from dataclasses import dataclass
from datetime import datetime, timezone
from decimal import Decimal
from typing import Callable, Protocol

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


@dataclass(frozen=True)
class ApplicationRecord:
    field_id: str
    product_epa_reg_no: str
    rate: Decimal          # regulated: Decimal, never float
    unit: str              # e.g. "L/ha"
    applied_at: datetime   # tz-aware UTC
    actor: str


class RuleSet(Protocol):
    version: str
    def validate(self, rec: ApplicationRecord) -> list[str]:
        """Return a list of violation codes; empty means the record conforms."""
        ...


class Ledger(Protocol):
    def append(self, record: dict, actor: str) -> object: ...
    def verify(self) -> tuple[bool, int | None]: ...


@dataclass(frozen=True)
class PipelineResult:
    written: int
    exceptions: list[tuple[str, list[str]]]  # (field_id, violation codes)
    manifest_hash: str
    submitted: bool


def run_seasonal_report(
    raw_payloads: list[dict],
    normalize: Callable[[dict], ApplicationRecord | None],
    rules: RuleSet,
    ledger: Ledger,
    render_package: Callable[[list[dict], str, str], bytes],
    *,
    season: str,
    jurisdiction: str = "EPA-FED",
    dry_run: bool = True,
) -> PipelineResult:
    """Telemetry -> normalized -> validated -> ledger -> rendered -> assembled."""
    exceptions: list[tuple[str, list[str]]] = []
    written = 0

    for raw in raw_payloads:
        # Stage 2 — normalize; a payload the parser cannot canonicalize is skipped,
        # not guessed, and its own upstream quarantine trail already records it.
        rec = normalize(raw)
        if rec is None:
            continue

        # Stage 3 — validate against the season's rule set.
        violations = rules.validate(rec)
        if violations:
            exceptions.append((rec.field_id, violations))
            logger.warning(
                '{"event":"record_rejected","field_id":"%s","codes":%s}',
                rec.field_id, json.dumps(violations),
            )
            continue  # exceptions never enter the compliant ledger slice

        # Stage 4 — write to the tamper-evident ledger, stamping the rule version.
        ledger.append(
            {
                "field_id": rec.field_id,
                "epa_reg_no": rec.product_epa_reg_no,
                "rate": str(rec.rate),
                "unit": rec.unit,
                "applied_at": rec.applied_at.astimezone(timezone.utc).isoformat(),
                "rule_version": rules.version,
            },
            actor=rec.actor,
        )
        written += 1

    # Integrity gate: never render a package from an unverifiable ledger.
    ok, broken = ledger.verify()
    if not ok:
        logger.error('{"event":"pipeline_halted","reason":"chain_break","index":%s}', broken)
        raise RuntimeError(f"ledger chain broken at index {broken}; refusing to file")

    # Stage 5 + 6 — render the deterministic package and hash the manifest.
    package_bytes = render_package(
        [json.loads(json.dumps(e)) for e in _slice_for(season, jurisdiction)],
        season, jurisdiction,
    )
    manifest_hash = hashlib.sha256(package_bytes).hexdigest()
    logger.info(
        '{"event":"package_assembled","season":"%s","jurisdiction":"%s",'
        '"written":%d,"exceptions":%d,"manifest_hash":"%s","dry_run":%s}',
        season, jurisdiction, written, len(exceptions),
        manifest_hash[:12], str(dry_run).lower(),
    )
    return PipelineResult(
        written=written, exceptions=exceptions,
        manifest_hash=manifest_hash, submitted=not dry_run,
    )


def _slice_for(season: str, jurisdiction: str) -> list[dict]:
    """Ledger slice selection is owned by the export guide; stubbed here for shape."""
    return []

The ordering is deliberate: validation happens before the write, so the ledger only ever holds records that conformed to the rule version stamped on them, and verify() runs before rendering, so a package is never built from a chain that cannot be proven intact.

Reruns are the norm, not the exception — a season is rarely closed in a single pass, and a partial run must not corrupt the record. Two properties make reruns safe. First, appends are idempotent at the caller: each ApplicationRecord carries a business key (field_id plus applied_at plus product), and the orchestrator deduplicates against the existing ledger slice before appending, so re-processing the same telemetry never writes a second copy of an application. Second, the exceptions annex is rebuilt from scratch on every run rather than accumulated, so a record that was rejected under a stale rule_version and later passes after the version is corrected simply moves from the annex into the compliant slice on the next pass, with no manual cleanup. The integrity gate then guarantees that whatever slice is rendered was verifiable at render time, regardless of how many partial runs preceded it.

Step-by-step procedure

1. Collect the raw telemetry for the season. Pull the ISOBUS and OEM API payloads that fall inside the reporting window. These arrive already fetched and quarantine-triaged by Equipment Telemetry Parsing; this step only scopes them to the season.
2. Normalize each payload to a canonical record. Run every payload through normalize, producing an ApplicationRecord with a Decimal rate and a tz-aware UTC timestamp. Payloads the parser cannot canonicalize are skipped here — their upstream quarantine trail already accounts for them.
3. Validate against the season’s rule set. Check each record with the pinned rule_version. A record with any violation code is routed to the exceptions annex for manual review and never enters the compliant record set.
4. Write conforming records to the audit ledger. Append each valid record, stamping the rule version so the entry re-verifies against the contract in force at application time. The hash chain makes any later edit detectable.
5. Verify the chain before rendering. Run verify(); if it reports a break, halt and file nothing. A package built on an unverifiable ledger is worse than no package.
6. Render the statutory PDF and CSV. Render the season’s ledger slice into the jurisdiction template deterministically, so the output is byte-stable and reconcilable against any later regeneration.
7. Assemble and hash the submission package. Bundle the PDF, the CSV, and a manifest into the EPA package, compute the manifest hash, and record it. With dry_run=True the package is produced for review; setting it false marks the filing submitted.

Log patterns and observable signals

Success path (season closed cleanly):

{"event": "package_assembled", "season": "2026", "jurisdiction": "EPA-FED", "written": 1284, "exceptions": 3, "manifest_hash": "b71f0c9a44de", "dry_run": true}

A record failed rule validation and was routed to the annex:

{"event": "record_rejected", "field_id": "north-40", "codes": ["REI_WINDOW_VIOLATION", "RATE_ABOVE_LABEL_MAX"]}

The ledger chain did not verify — the run halted before any file was produced:

{"event": "pipeline_halted", "reason": "chain_break", "index": 842}

Triage pipeline_halted first: it means no package was filed and the ledger integrity must be restored from backup before a rerun. A rising record_rejected rate for one product usually points at a stale rule_version or a normalization drift upstream, not a genuine field violation.

Safe override protocol

Occasionally a season must be filed with a record that failed automated validation but is legitimate on review — a label rate that a supplemental EPA registration permits, for example. The override must never inject the record straight into the compliant set; it only re-classifies a specific, named exception after human sign-off.

Guard conditions, all mandatory:

1. Named record, never a blanket pass. The override targets one field_id plus violation code, never “accept all exceptions.”
2. Documented authority. The reviewer supplies the EPA registration or supplemental-label reference that makes the record admissible, and it is stamped into the entry.
3. Dry-run reconciliation. The overridden package is rendered with dry_run=True and the reviewer confirms the record now appears correctly before submission.
4. Immutable trail. The original violation, the override reason, and the approver identity are appended to the ledger as their own entry, consistent with EPA pesticide recordkeeping expectations, so the exception and its justification are both permanent.

Troubleshooting

• pipeline_halted with reason: chain_break. Root cause: a historical ledger entry was altered, so verify() failed before rendering. Remediation: restore the ledger from the last verified off-site backup; never re-hash in place to silence the alarm, and investigate the write path that mutated an entry.
• Filed rates disagree with the sprayer readout by a rounding step. Root cause: a rate was coerced to float somewhere before the Decimal boundary. Remediation: keep rates as Decimal from normalization through the ledger; raise min_rate_precision only if the source genuinely carries more places.
• A whole product’s records land in the exceptions annex. Root cause: the pinned rule_version does not match the season, so valid applications trip a stale threshold. Remediation: confirm the season’s rule_version against EPA/USDA Rule Mapping before the run; re-validate after correcting it.
• The regenerated package hash differs from the filed one. Root cause: a non-deterministic element (embedded timestamp, unsorted rows) leaked into rendering. Remediation: render from the canonical ledger slice with sorted keys and no wall-clock metadata, so the manifest hash is reproducible.
• Normalized count is far below the raw payload count. Root cause: a firmware schema drift is causing the parser to skip payloads. Remediation: inspect the upstream quarantine volume in Equipment Telemetry Parsing rather than loosening validation here.

Frequently asked questions

Why validate before writing to the ledger? So the ledger holds only records that conformed to the rule version stamped on them; violations go to a separate annex and never mix into the compliant slice that gets filed.

What if verify() fails mid-run? The pipeline halts and files nothing. Restore the ledger from a verified backup and investigate the write path — never re-hash in place to quiet the alarm.

How are the right rules applied to the right season? The rule_version is pinned per season and stamped into each entry, so records re-validate against the contract in force at application time; a season mismatch surfaces as an oversized exceptions annex.

Related

• Equipment Telemetry Parsing — normalizes the raw machine payloads this pipeline consumes at stage 2.
• EPA/USDA Rule Mapping — the versioned rule set every record is validated against at stage 3.
• Generating Statutory PDF and CSV Export Packages — the deterministic rendering used at stages 5 and 6.

Up: Audit & Reporting · Agricultural Automation System Architecture & Compliance