Conflict Resolution in Multi-Agency Edits: Python GIS Workflows

A fire branch redraws an evacuation perimeter at 14:02. A sheriff’s deputy, still on a degraded cellular link, syncs an hour-old replica at 14:47 that reopens a structure the fire branch already closed. If the synchronization layer applies a naive last-writer-wins merge, the closed structure resurrects as “active,” units are committed to stale geometry, and the after-action review (AAR) inherits a timeline that cannot be reconciled. That single overwrite is the operational failure this workflow exists to prevent: deterministic, auditable reconciliation of concurrent edits across jurisdictions that never silently discards an authoritative change. The wider architecture this slots into is defined in Incident Mapping & Multi-Agency Sync Workflows, where versioned feature services, local replicas, and a central reconciliation engine operate under strict latency and data-integrity constraints.

Prerequisites

This pattern assumes the upstream pipeline is already in place; it reconciles deltas, it does not clean them.

• Python packages: geopandas >= 0.14, shapely >= 2.0 (for make_valid and the vectorised predicate API), pandas, and sqlite3 from the standard library for offline replay. pyproj is pulled in transitively for reprojection.
• CRS contract: every delta arrives in EPSG:4326 for storage. Area and distance comparisons happen only after reprojection to a metric CRS (an appropriate UTM zone, or EPSG:3857 for coarse display thresholds). Mixed-axis-order payloads must already be normalised — that normalisation is owned by Real-Time Geocoding & Location Normalization, and feeding un-normalised coordinates into the resolver produces false-positive overlap flags from projection drift.
• Schema contract: every feature carries a stable, globally unique merge key (incident_id), an agency_type that maps to an Incident Command System (ICS) authority tier, and a monotonic last_edited timestamp. Records that fail the field contract should be rejected by Automated Attribute Validation Rules before they ever reach this stage.

Conflict Taxonomy and Resolution Flow

Edit conflicts in tactical GIS manifest along three axes: spatial topology violations (intersecting evacuation zones, overlapping hazard perimeters), attribute divergence (conflicting severity, status, or resource counts on the same incident_id), and temporal sequencing mismatches (an offline replica replaying older edits over newer ones). The resolver treats these in a fixed order — repair geometry, classify, then resolve — so the outcome is reproducible regardless of which replica synced first.

Step-by-Step Implementation

The service follows a staged execution model — ingestion, validation/classification, resolution, commit — and each stage is stateless so it can run in a container and be horizontally scaled behind the ingestion queue.

1. Delta ingestion and version tracking

The reconciliation engine extracts created, updated, and deleted records from each agency replica and tags every delta with provenance before any merge logic runs. Concurrent sync requests are handled with asyncio; the snippet below isolates the tagging contract that the later stages depend on.

import logging
from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Literal

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(name)s - %(message)s",
)
logger = logging.getLogger("conflict.ingest")

DeltaOp = Literal["created", "updated", "deleted"]


@dataclass(frozen=True)
class TaggedDelta:
    incident_id: str
    op: DeltaOp
    source_agency: str
    agency_type: str            # FEDERAL | STATE | LOCAL | VOLUNTEER
    last_edited: datetime
    edit_sequence: int
    payload: dict = field(default_factory=dict)


def tag_delta(raw: dict, source_agency: str, agency_type: str) -> TaggedDelta:
    """Attach provenance to a raw edit. Rejects payloads missing the merge key."""
    try:
        return TaggedDelta(
            incident_id=raw["incident_id"],
            op=raw["op"],
            source_agency=source_agency,
            agency_type=agency_type,
            last_edited=datetime.fromisoformat(raw["last_edited"]).astimezone(timezone.utc),
            edit_sequence=int(raw["edit_sequence"]),
            payload=raw.get("payload", {}),
        )
    except (KeyError, ValueError) as exc:
        logger.error("Dropping untaggable delta from %s: %s", source_agency, exc)
        raise

2. Validation and conflict classification

Geometry is repaired before any spatial predicate runs — an invalid self-intersecting polygon will raise inside overlay() and abort the whole batch otherwise. The classifier returns explicit conflict reports rather than mutating state, so resolution and commit stay separable and testable.

import geopandas as gpd
from shapely.geometry import shape
from shapely.validation import make_valid
from typing import Optional

logger = logging.getLogger("conflict.classify")


def validate_geometry(geom_dict: dict) -> Optional[gpd.GeoSeries]:
    """Repair invalid geometry; return None (and log) on unrecoverable input."""
    try:
        geom = shape(geom_dict)
        if not geom.is_valid:
            logger.warning("Repairing invalid geometry via make_valid")
            geom = make_valid(geom)
        return gpd.GeoSeries([geom], crs="EPSG:4326")
    except Exception as exc:  # noqa: BLE001 - geometry libs raise a wide set
        logger.error("Geometry validation failed: %s", exc)
        return None


def classify_spatial_overlap(
    gdf_a: gpd.GeoDataFrame,
    gdf_b: gpd.GeoDataFrame,
    crs_metric: str = "EPSG:3857",
) -> list[dict]:
    """
    Report intersecting features. Area is measured in a metric CRS, never in
    EPSG:4326 degrees, so overlap thresholds are expressed in square metres.
    """
    conflicts: list[dict] = []
    try:
        a_m = gdf_a.to_crs(crs_metric)
        b_m = gdf_b.to_crs(crs_metric)
        intersections = gpd.overlay(a_m, b_m, how="intersection")
        for _, row in intersections.iterrows():
            conflicts.append({
                "feature_ids": [row.get("id_1"), row.get("id_2")],
                "overlap_area_m2": float(row.geometry.area),
                "same_merge_key": row.get("id_1") == row.get("id_2"),
                "resolution_status": "PENDING_REVIEW",
            })
    except Exception as exc:  # noqa: BLE001
        logger.error("Spatial overlay failed: %s", exc)
    return conflicts

3. Precedence-weighted attribute resolution

Attribute conflicts on a shared incident_id resolve on ICS authority first; last-writer-wins is the tiebreaker only, never the primary rule. This is the inversion of naive sync: a fresh VOLUNTEER edit must never overwrite an older FEDERAL one.

logger = logging.getLogger("conflict.resolve")

AGENCY_PRECEDENCE: dict[str, int] = {
    "FEDERAL": 3,
    "STATE": 2,
    "LOCAL": 1,
    "VOLUNTEER": 0,
}


def resolve_attribute_conflict(
    base: TaggedDelta,
    delta: TaggedDelta,
    precedence: dict[str, int] = AGENCY_PRECEDENCE,
) -> TaggedDelta:
    """Authority precedence wins; equal authority falls back to last-writer-wins."""
    try:
        base_rank = precedence[base.agency_type]
        delta_rank = precedence[delta.agency_type]
    except KeyError as exc:
        logger.critical("Unknown agency_type in precedence map: %s", exc)
        raise ValueError("Incomplete payload for conflict evaluation") from exc

    if delta_rank != base_rank:
        winner = delta if delta_rank > base_rank else base
        logger.info(
            "Resolved %s by authority: %s (%s) over %s",
            base.incident_id, winner.source_agency, winner.agency_type,
            base.source_agency if winner is delta else delta.source_agency,
        )
        return winner

    winner = delta if delta.last_edited >= base.last_edited else base
    logger.info("Resolved %s by last-writer-wins: %s", base.incident_id, winner.source_agency)
    return winner

4. Offline replay and commit with audit emission

Field operations run in degraded-network environments, so the resolver must replay edits cached during an outage and reconcile them against the canonical state on reconnect. Local replicas use SQLite; the SQLite-backed storage and delta-extraction patterns are covered in depth by Syncing ArcGIS Online edits to local GeoPackage. Every committed resolution emits an immutable audit row.

import json
import sqlite3

logger = logging.getLogger("conflict.commit")


def commit_resolution(conn: sqlite3.Connection, base: TaggedDelta, winner: TaggedDelta) -> None:
    """Apply the reconciled state and append an immutable audit record atomically."""
    rule = "authority" if winner.agency_type != base.agency_type else "last_writer_wins"
    try:
        with conn:  # transactional: feature update + audit insert commit together
            conn.execute(
                "UPDATE cop_features SET payload = ?, last_edited = ?, source_agency = ? "
                "WHERE incident_id = ?",
                (json.dumps(winner.payload), winner.last_edited.isoformat(),
                 winner.source_agency, winner.incident_id),
            )
            conn.execute(
                "INSERT INTO audit_log "
                "(incident_id, original_state, delta_state, resolution_rule, ts_utc) "
                "VALUES (?, ?, ?, ?, ?)",
                (base.incident_id, json.dumps(base.payload), json.dumps(winner.payload),
                 rule, datetime.now(timezone.utc).isoformat()),
            )
    except sqlite3.Error as exc:
        logger.error("Commit failed for %s, rolling back: %s", base.incident_id, exc)
        raise

Unresolvable spatial overlaps between distinct features (where same_merge_key is false) are never auto-merged. They route to a conflict_quarantine table with the full payload preserved for adjudication by a GIS supervisor — automatically collapsing two real perimeters can hide a resource gap, which is exactly the failure class this workflow guards against.

Configuration Reference

These parameters are the tunable surface of the resolver. Treat them as policy, version-controlled alongside the service, not as inline constants.

Verification and Smoke Test

Before promoting a resolver build to staging, assert the two invariants that field incidents actually depend on: authority must beat recency, and area thresholds must be evaluated in metres. The block below is runnable against the functions above with no external service.

from datetime import timedelta


def _smoke() -> None:
    now = datetime.now(timezone.utc)
    base = TaggedDelta("INC-001", "updated", "USFS", "FEDERAL", now, 10, {"status": "closed"})
    # A NEWER volunteer edit must NOT win over an older federal edit.
    stale_authority_vs_fresh_volunteer = TaggedDelta(
        "INC-001", "updated", "CERT", "VOLUNTEER", now + timedelta(minutes=45), 11,
        {"status": "active"},
    )
    winner = resolve_attribute_conflict(base, stale_authority_vs_fresh_volunteer)
    assert winner is base, "authority precedence must override recency"
    assert winner.payload["status"] == "closed"
    logger.info("smoke ok: authority precedence holds")


if __name__ == "__main__":
    _smoke()

Run it as a CI gate:

python -m conflict.resolver && echo "resolver smoke PASS"

Integration with Adjacent Workflows

The resolver is the reconciliation stage of a longer pipeline and depends on its neighbours holding their contracts. Coordinates must be canonicalised upstream by Real-Time Geocoding & Location Normalization so that overlap classification does not fire on projection drift, and the field-level schema must already be enforced by Automated Attribute Validation Rules so the merge key and agency_type are guaranteed present. High-frequency edit streams from sensors, UAV feeds, and mobile CAD terminals are decoupled from reconciliation by WebSocket & MQTT for Live Incident Feeds, which buffers deltas into a queue and feeds the resolver in micro-batches to prevent head-of-line blocking on the command console.

Troubleshooting

Symptom: gpd.overlay raises TopologyException mid-batch. Root cause: an invalid input geometry slipped past validation. Ensure every feature passes through validate_geometry before it enters classify_spatial_overlap; make_valid resolves self-intersections and ring-order errors that abort the C-level predicate.

Symptom: overlap areas look absurdly small (e.g. 0.0003). Root cause: area was computed in EPSG:4326, so the units are square degrees. Reproject to a metric CRS first — classify_spatial_overlap calls to_crs for exactly this reason; verify no caller bypasses it.

Symptom: a closed feature reopens after a replica reconnects. Root cause: the merge fell through to last-writer-wins because both records carried the same agency_type, or the offline replica’s clock was ahead. Confirm agency_type is populated (not defaulted) and that last_edited is UTC; widen RESOLVER_SKEW_S only after auditing the device clocks.

Symptom: distinct perimeters silently collapse into one feature. Root cause: a quarantine path was bypassed and two features sharing an overlap but not a merge key were auto-merged. Assert same_merge_key before merging; route everything else to conflict_quarantine.

Symptom: audit rows are missing for some commits. Root cause: the feature update and audit insert were not in one transaction, so a mid-commit error left the update applied but the audit row absent. Keep both writes inside the single with conn: block in commit_resolution so they commit or roll back together.

Related

• Real-Time Geocoding & Location Normalization — canonicalise coordinates before reconciliation
• Automated Attribute Validation Rules — enforce the merge-key and field contract upstream
• WebSocket & MQTT for Live Incident Feeds — decouple high-frequency edit streams from the resolver
• Syncing ArcGIS Online edits to local GeoPackage — offline replica storage and delta extraction

Up: Incident Mapping & Multi-Agency Sync Workflows