Field Boundary Synchronization

Field boundary synchronization is the control plane of the ingestion platform: it decides which polygon is the truth for a field when a dozen sources disagree about where that field’s edges are. An RTK guidance trace, a co-op’s desktop GIS export, an agronomist’s hand-drawn correction, and last season’s stored boundary can all describe the same parcel and still differ by metres of acreage. This subsystem sits inside the Farm Data Ingestion & Field Boundary Synchronization pipeline and owns one job: reconcile every incoming geometry against a single canonical registry, resolve the conflict deterministically, version the outcome, and replay a backlog identically after a rural connection drops and returns.

It is important to separate this responsibility from two neighbours it is easily confused with. Equipment Telemetry Parsing decodes machine payloads into events and hands off validated geometry; the boundary-file decoder in Parsing ISOXML and Shapefile field boundaries turns a TASKDATA.XML or a shapefile into a clean WGS84 polygon. Both produce candidate geometry. This subsystem is what happens next: it does not decode payloads and it does not enforce coordinate reference systems — it takes already-decoded candidates and answers whether each one should become the new canonical shape of the field, stay quarantined, or be discarded as a duplicate.

Problem framing: many sources, one truth, and a network that comes and goes

A field’s boundary is not a fixed fact; it is a running claim that shifts as tile drainage is installed, a headland is re-cut, a waterway easement is surveyed, or a machine simply logs a sloppier pass than last week. Each of those events arrives as a slightly different polygon, and the platform cannot treat the newest arrival as automatically correct. If it did, one uncorrected GPS wander during a foggy morning would silently shrink a field’s recorded acreage, and every rate calculation, every USDA Farm Service Agency acreage report, and every input-tracking total keyed to that field would inherit the error without a trace.

The specific sub-problem here is reconciliation under three simultaneous pressures. First, sources have different trust: a survey-grade operator edit is worth more than an RTK trace, which is worth more than a bulk GIS export whose provenance is unknown. Second, changes have different magnitudes: a two-centimetre jitter is noise, a two-hectare change is a real event, and the system must tell them apart without a human in the loop for the common case. Third, connectivity is intermittent: a sprayer records boundary observations for hours in a dead zone, then reconnects and floods the ingestion loop with a backlog that must be applied in a deterministic order, or two worker nodes draining the same buffer will produce two different registries.

The failure mode this subsystem exists to prevent is a quiet fork: two divergent “current” boundaries for one field, each blessed by a different worker, each feeding a different downstream calculation. The remedy is a single canonical registry with an append-only revision chain, a deterministic reconcile path, and a hard rule that geometry large enough to matter is held for review rather than auto-committed. The cadence question — when candidates are pulled — belongs to the Async Polling Strategies loop; this subsystem owns what the registry does with each candidate once it arrives.

Prerequisites and dependencies

The reconcile service sits downstream of decoding and upstream of every spatial calculation, so it depends on three upstream contracts and a small pinned stack. Confirm each before wiring it into ingestion.

• Decoded, CRS-defined candidate geometry. This service assumes every incoming polygon is already valid and in a known projection. Coordinate-system enforcement, ring winding, and multipart repair are the job of Parsing ISOXML and Shapefile field boundaries — never re-implemented here.
• A canonical event source. Candidates are surfaced by the Equipment Telemetry Parsing subsystem and delivered on the cadence set by Async Polling Strategies, including the buffered backlog that arrives on reconnect.
• A governed field record and threshold set. The canonical field record the registry writes into is defined by Field Schema Design, and when reconciliation cannot resolve a conflict automatically it escalates through the same conservative path as Fallback Routing Logic rather than guessing.

Library versions are pinned for reproducibility across worker nodes: shapely>=2.0 (the vectorized hausdorff_distance and make_valid used by the diff strategy), pydantic>=2.5 for frozen, validated revision records, and Python 3.10+ for match/case and X | None unions. All geometry is stored in an equal-area projection such as USA Contiguous Albers (EPSG:5070) so that area deltas and setback distances are in real metres, and acreage reconciled against the USDA Farm Service Agency is exact. Spatial storage conventions follow USDA NRCS guidance for common land units, and all revision timestamps are tz-aware UTC.

Architecture of this subsystem

The service is a reconcile engine wrapped around an append-only registry. A candidate geometry is validated and hashed, compared against the field’s active revision by an injectable difference strategy, resolved against source precedence when the change is contested, and then either minted as a new revision, held for an operator, discarded as an idempotent replay, or rejected outright. Every outcome is a structured log line carrying the field id, and every accepted change becomes an immutable link in that field’s revision chain — the previous shape is never edited in place, only superseded.

The inputs are an IncomingBoundary (a candidate geometry plus its source and observation time) and the field’s current registry state; the outputs are a SyncDecision and, when a change is accepted, a freshly appended BoundaryRevision. The integration is one-directional: the service consumes candidates and produces revisions that the acreage-reporting and audit layers of the parent section read — it never mutates the polling loop, and it never overwrites a stored revision.

Step-by-step implementation

The reference service below is a single reconcile class over an append-only registry, with an injectable difference strategy so the comparison math can evolve without touching the versioning logic. It runs on Python 3.10+ and passes a syntax and indentation check.

Step 1 — Model an immutable revision and an append-only registry

A revision is frozen so it cannot be edited after it is written; the only way to change a field is to append a successor whose parent_hash points at the current head. The registry enforces that chain, rejecting any write whose parent is not the current head — the optimistic-concurrency guard that stops two workers from forking one field.

Step 2 — Reconcile a candidate against the active revision

Reconciliation validates the geometry, hashes it canonically, and short-circuits an identical replay as an idempotent no-op. A genuine change is classified by the injected differ; a change large enough to need review is held, and a lower-ranked source is additionally logged as a deferred conflict so it can never quietly overwrite a higher-ranked shape.

Step 3 — Drain a reconnect backlog deterministically

When a machine reconnects after hours offline, sync_on_reconnect sorts the buffered candidates by observation time, then source rank, then field id, and applies them through the same reconcile path. The deterministic sort is what guarantees that any worker replaying the same buffer converges on the same registry state.

from __future__ import annotations

import hashlib
import json
import logging
from collections.abc import Iterable
from datetime import datetime, timezone
from decimal import Decimal
from enum import Enum
from typing import Protocol

from pydantic import BaseModel, ConfigDict
from shapely.geometry import mapping, shape
from shapely.geometry.base import BaseGeometry
from shapely.validation import make_valid

logger = logging.getLogger("boundary.sync")

_M2_PER_HA = Decimal("10000")
# Source precedence: a surveyed operator edit outranks an RTK trace, which in
# turn outranks a third-party GIS export. A lower-ranked source may not silently
# overwrite a change large enough to need review.
_SOURCE_RANK = {"operator_edit": 3, "rtk_trace": 2, "gis_export": 1}


class SyncDecision(str, Enum):
    UNCHANGED = "unchanged"
    AUTO_ACCEPTED = "auto_accepted"
    HELD_FOR_REVIEW = "held_for_review"
    REJECTED = "rejected"


class BoundaryRevision(BaseModel):
    """One immutable version of a field polygon; superseded, never edited."""

    model_config = ConfigDict(frozen=True)

    field_id: str
    revision: int
    geojson: dict
    area_ha: Decimal
    source: str
    geom_hash: str
    parent_hash: str | None
    recorded_at: datetime


class IncomingBoundary(BaseModel):
    """A candidate geometry arriving from a trace or export, pre-reconcile."""

    model_config = ConfigDict(arbitrary_types_allowed=True)

    field_id: str
    geom: BaseGeometry
    source: str
    observed_at: datetime


class GeometryDiffer(Protocol):
    """Injectable comparison strategy; the change-detection guide supplies the
    full IoU/Hausdorff implementation used in production."""

    def classify(self, current: BaseGeometry, candidate: BaseGeometry) -> SyncDecision:
        ...


def _canonical_hash(geojson: dict) -> str:
    """Stable SHA-256 over sorted GeoJSON so identical geometry hashes identically."""
    blob = json.dumps(geojson, sort_keys=True, separators=(",", ":"))
    return hashlib.sha256(blob.encode("utf-8")).hexdigest()


def _area_ha(geom: BaseGeometry) -> Decimal:
    # Geometry is stored in an equal-area projection (e.g. EPSG:5070), so planar
    # area is metric; Decimal keeps FSA acreage reconciliation exact.
    return (Decimal(str(geom.area)) / _M2_PER_HA).quantize(Decimal("0.0001"))


class CanonicalRegistry:
    """Append-only store of BoundaryRevisions keyed by field_id."""

    def __init__(self) -> None:
        self._chains: dict[str, list[BoundaryRevision]] = {}

    def active(self, field_id: str) -> BoundaryRevision | None:
        chain = self._chains.get(field_id)
        return chain[-1] if chain else None

    def history(self, field_id: str) -> list[BoundaryRevision]:
        return list(self._chains.get(field_id, ()))

    def append(self, revision: BoundaryRevision) -> BoundaryRevision:
        chain = self._chains.setdefault(revision.field_id, [])
        expected_parent = chain[-1].geom_hash if chain else None
        # Optimistic-concurrency guard: reject a write whose parent is not the
        # current head, so two workers cannot fork the same chain.
        if revision.parent_hash != expected_parent:
            raise ValueError(
                f"{revision.field_id}: stale parent_hash; expected {expected_parent}"
            )
        if revision.revision != len(chain) + 1:
            raise ValueError(f"{revision.field_id}: non-monotonic revision number")
        chain.append(revision)
        return revision


class BoundarySyncService:
    """Reconciles incoming geometry against the canonical registry and versions it."""

    def __init__(self, registry: CanonicalRegistry, differ: GeometryDiffer) -> None:
        self._registry = registry
        self._differ = differ

    def reconcile(
        self, incoming: IncomingBoundary
    ) -> tuple[SyncDecision, BoundaryRevision | None]:
        geom = make_valid(incoming.geom)
        if geom.is_empty or geom.geom_type not in ("Polygon", "MultiPolygon"):
            logger.error(json.dumps(
                {"event": "sync_rejected", "field_id": incoming.field_id,
                 "reason": "non_areal_geometry"}))
            return SyncDecision.REJECTED, self._registry.active(incoming.field_id)

        geojson = mapping(geom)
        incoming_hash = _canonical_hash(geojson)
        active = self._registry.active(incoming.field_id)

        # Idempotency: a byte-identical geometry replayed on reconnect is a no-op.
        if active is not None and active.geom_hash == incoming_hash:
            logger.info(json.dumps(
                {"event": "sync_idempotent", "field_id": incoming.field_id,
                 "revision": active.revision}))
            return SyncDecision.UNCHANGED, active

        if active is None:
            decision = SyncDecision.AUTO_ACCEPTED  # first observation seeds the chain
        else:
            decision = self._differ.classify(shape(active.geojson), geom)
            if decision is SyncDecision.HELD_FOR_REVIEW:
                if _SOURCE_RANK.get(incoming.source, 0) < _SOURCE_RANK.get(active.source, 0):
                    logger.warning(json.dumps(
                        {"event": "sync_conflict_deferred",
                         "field_id": incoming.field_id,
                         "incoming_source": incoming.source,
                         "active_source": active.source}))
                logger.warning(json.dumps(
                    {"event": "sync_held_for_review", "field_id": incoming.field_id}))
                return SyncDecision.HELD_FOR_REVIEW, active
            if decision is SyncDecision.REJECTED:
                logger.error(json.dumps(
                    {"event": "sync_rejected", "field_id": incoming.field_id,
                     "reason": "differ_rejected"}))
                return SyncDecision.REJECTED, active

        revision = BoundaryRevision(
            field_id=incoming.field_id,
            revision=(active.revision + 1) if active else 1,
            geojson=geojson,
            area_ha=_area_ha(geom),
            source=incoming.source,
            geom_hash=incoming_hash,
            parent_hash=active.geom_hash if active else None,
            recorded_at=datetime.now(timezone.utc),
        )
        stored = self._registry.append(revision)
        logger.info(json.dumps(
            {"event": "revision_minted", "field_id": stored.field_id,
             "revision": stored.revision, "geom_hash": stored.geom_hash[:12],
             "area_ha": str(stored.area_ha), "decision": decision.value}))
        return decision, stored

    def sync_on_reconnect(
        self, buffered: Iterable[IncomingBoundary]
    ) -> dict[str, SyncDecision]:
        """Drain a disconnected backlog deterministically: order by observed time,
        then source rank, then field_id, so replaying the same buffer always
        produces the same registry state."""
        ordered = sorted(
            buffered,
            key=lambda b: (b.observed_at, -_SOURCE_RANK.get(b.source, 0), b.field_id),
        )
        outcomes: dict[str, SyncDecision] = {}
        for item in ordered:
            decision, _ = self.reconcile(item)
            outcomes[item.field_id] = decision
        return outcomes

Expected log output when a real change from a trusted source is accepted and versioned:

{"event": "revision_minted", "field_id": "CLU-114", "revision": 4, "geom_hash": "9f2c41ab77e0", "area_ha": "32.6180", "decision": "auto_accepted"}

Expected log output when a lower-ranked GIS export contests a boundary the operator already fixed:

{"event": "sync_conflict_deferred", "field_id": "CLU-114", "incoming_source": "gis_export", "active_source": "operator_edit"}
{"event": "sync_held_for_review", "field_id": "CLU-114"}

The history accessor deliberately returns a copy of the chain: an auditor can walk every prior shape of a field, but nothing outside the registry can splice or truncate the list.

Edge cases and known failure modes

Compliance and audit integration

The registry is a compliance artifact before it is a convenience. A USDA Farm Service Agency acreage report is only defensible if the polygon it was computed from can be produced exactly as it stood on the report date — which is precisely what the append-only chain guarantees. Because a revision is superseded rather than edited, an acreage figure filed in June replays against the revision that was active in June, even after a July re-survey changes the field. The area_ha field is carried as a Decimal so the reconciliation against FSA common-land-unit acreage is exact to the hundredth of a hectare rather than drifting through float rounding.

Three facts must survive from reconcile to inspection: the geometry hash of every revision, the source and decision that produced it, and the parent link that orders the chain. When reconciliation cannot resolve a conflict — a low-trust source contesting a surveyed edit, or a change beyond the auto-accept envelope — it escalates conservatively through the same discipline as Fallback Routing Logic: hold, log, and wait for a human rather than commit a guess. Spatial conventions follow USDA NRCS practice for common land units, and the canonical record shape is governed centrally by Field Schema Design so a schema change never rewrites a historical boundary. Structured logs should stream to a centralized observability platform so held-for-review volume, conflict-deferral rate, and revision churn per field are monitored as health signals — a sudden spike in sync_conflict_deferred usually means a bad bulk import is fighting the registry.

Verification

Confirm correct operation in staging with a reproducible fault-injection scenario before promoting a change:

1. Seed and replay. Reconcile a first geometry, assert it mints revision 1, then reconcile the byte-identical geometry again and assert the result is UNCHANGED with no new revision — the idempotency guarantee that makes reconnect safe.
2. Inject a trivial jitter. Feed a version perturbed below the differ’s overlap threshold and assert it is reported UNCHANGED, so RTK noise never floods the chain.
3. Inject a contested change. With an operator_edit active, reconcile a materially different gis_export and assert it is HELD_FOR_REVIEW and that sync_conflict_deferred is logged — never auto-committed over the surveyed shape.
4. Force a concurrency fork. Append a revision, then attempt a second append carrying the previous head as its parent_hash, and assert append raises on the stale parent rather than forking the chain.
5. Shuffle a backlog. Pass sync_on_reconnect the same buffer in two different arrival orders and assert both runs leave an identical registry, proving the deterministic drain.

A run passes only when each injected fault produces the expected conservative outcome and the chain contains exactly one traceable revision per accepted change. The detailed accept/quarantine math for individual GPS fixes and the diff thresholds behind the differ are covered in the two guides below.

Frequently Asked Questions

How is this different from parsing a boundary file? Boundary-file parsing decodes one archive into a valid polygon and stops there; it has no memory of the field. This subsystem is stateful — it compares each decoded candidate against the field’s canonical history, decides whether the change is real and trusted, and versions the outcome. Decoding produces a candidate; synchronization decides whether that candidate becomes the truth.

Why hold a change for review instead of always taking the newest geometry? Because the newest geometry is not always the most correct one. An uncorrected GPS pass or a stale bulk export can be newer than a survey-grade operator edit while being far less trustworthy. Holding a large or low-ranked change for review keeps a single noisy observation from silently rewriting the acreage every downstream calculation depends on.

What makes a reconnect backlog safe to replay? Two properties. Reconciliation is idempotent, so a geometry that arrives twice mints one revision; and sync_on_reconnect applies the buffer in a deterministic order — observation time, then source rank, then field id — so any worker draining the same backlog converges on the same registry rather than forking it.

How does the chain support an FSA acreage report? Every revision is immutable and carries its exact Decimal area and a hash. A report filed on a given date can be reproduced from the revision that was active on that date, even after later changes, and the parent-hash chain lets an auditor walk backward through every prior shape of the field without any record having been edited in place.

Related

• Reconciling GPS Drift Against Canonical Field Polygons — the accept/quarantine rule for individual near-edge fixes that feed this registry.
• Detecting and Versioning Field Boundary Changes — the IoU, area-delta, and Hausdorff diff behind the injected differ and the revision it mints.
• Equipment Telemetry Parsing — decodes the machine events that surface boundary candidates.
• Parsing ISOXML and Shapefile field boundaries — the decoder that produces the CRS-defined polygons this service reconciles.
• Async Polling Strategies — the fetch cadence and reconnect buffer that feed candidates in.
• Fallback Routing Logic — the conservative escalation pattern this subsystem follows when a conflict cannot auto-resolve.

Up: Farm Data Ingestion & Field Boundary Synchronization