Stamping source-to-target lineage hashes on every record

An auditor sends one line: prove that parcel 048-217-009 in Pima County is zoned TR because of the June 2026 rezoning PDF you ingested, and not because of a mapping bug. You open the table, and the row has a zoning_code and a last_updated timestamp — and nothing else. No pointer to the PDF, no record of which extraction ran, no mapping version. The value might be correct, but you cannot prove it without re-scraping a portal whose contents have since changed. This is the retrofit problem: records already in production carry no provenance, and you need to start stamping every new record — and ideally reprocess the backlog — so that this question is answerable by lookup instead of archaeology. This guide is the hands-on companion to data lineage & provenance tracking within the broader Municipal Zoning Data Architecture & Compliance Frameworks area. It walks through canonicalizing the source, chaining a deterministic lineage hash, stamping it on the record, persisting an immutable audit row, and verifying that the whole thing reproduces bit-for-bit.

Diagnosis: what a record without provenance actually costs

Start by confirming the gap and reproducing what a reviewer would find. Query the row and inspect what evidence exists:

row = db.fetchone(
    "SELECT parcel_id, zoning_code, last_updated FROM parcels WHERE parcel_id = %s",
    ("048-217-009",),
)
print(row)
# {'parcel_id': '048-217-009', 'zoning_code': 'TR', 'last_updated': '2026-06-19'}

There are three distinct things missing, and each blocks a different part of the auditor’s question:

  1. No source binding. Nothing ties TR to a specific municipal publication. You have a date, but the portal’s June file may have been silently replaced, so even re-fetching proves nothing about what you actually ingested.
  2. No transform identity. Even with the right PDF, you cannot say which extraction code produced TR. If a regex changed between releases, the current code may not reproduce the historical value at all.
  3. No mapping version. The raw source code was almost certainly something local like Transitional-Res, translated to TR through a taxonomy table whose version is unrecorded — the same normalization concern handled upstream by attribute normalization rules. Without the version, a reviewer cannot tell whether TR reflects the mapping in force in June or one edited since.

The fix is to compute a single lineage token that binds all three inputs, stamp it on the record, and store an immutable audit row plus the canonical source bytes. The flow looks like this:

Stamping a source-to-target lineage hash onto one zoning record The original source bytes are canonicalized and hashed into a source content address. The transform id and mapping version are captured as inputs. The source address, transform id, and mapping version are chained into a deterministic lineage hash. That token is stamped onto the parcel record and, in parallel, written to an append-only audit row and a content-addressed manifest of the source bytes. A verification step re-runs the chain and asserts the recomputed token is identical. re-run → identical token? Source bytes canonical → sha256 Transform id extract@version Mapping version taxonomy@date Chain → lineage hash deterministic token Stamp on record zoning_code + lin: Append-only audit row Content-addressed manifest of source

Step-by-step implementation

Each step handles one concern. The final step composes them into a single stamping call you can run per record or in a backfill loop.

Step 1 — Canonicalize the source bytes and hash them

The source address must be reproducible, so reduce the input to a fixed byte form before hashing. Hash the original file for PDFs; only serialize structured payloads with sorted keys and no incidental whitespace.

import hashlib
import json
from typing import Any


def content_address(payload: Any) -> str:
    """Stable sha256 address of a canonicalized payload."""
    if isinstance(payload, bytes):
        canon = payload                      # raw PDF/API bytes hashed as-is
    else:
        canon = json.dumps(
            payload, sort_keys=True, separators=(",", ":"), ensure_ascii=False
        ).encode("utf-8")
    return "sha256:" + hashlib.sha256(canon).hexdigest()


src_addr = content_address(open("pima_rezone_2026-06.pdf", "rb").read())

Step 2 — Capture the transform inputs

The transform identity is a durable code version plus a logical step name, and the mapping version comes from whatever taxonomy table translated the raw code. Read both from configuration or the package metadata rather than hardcoding a literal, so a release bump automatically flows into the hash.

from importlib import metadata


def transform_identity(step: str, package: str = "zoning_extract") -> str:
    """e.g. '[email protected]/rezoning-pdf'."""
    try:
        version = metadata.version(package)
    except metadata.PackageNotFoundError:
        version = "0.0.0-dev"                 # never silently omit the version
    return f"{package}@{version}/{step}"


transform_id = transform_identity("rezoning-pdf")
mapping_version = "[email protected]"    # from the mapping table's own hash

Step 3 — Compute the chained lineage hash

Join the three inputs with a NUL delimiter so different splits of the same characters can never collide, prefix with an algorithm version, and hash. This is the single token that binds source, transform, and mapping.

ALGO = "lineage-v1"


def lineage_hash(source_address: str, transform_id: str, mapping_version: str) -> str:
    if not (source_address and transform_id and mapping_version):
        raise ValueError("all three lineage inputs are required")
    joined = "\x00".join([ALGO, source_address, transform_id, mapping_version])
    return "lin:" + hashlib.sha256(joined.encode("utf-8")).hexdigest()


token = lineage_hash(src_addr, transform_id, mapping_version)

Step 4 — Stamp the token on the record

Add a lineage_id column if it does not exist, then write the token beside the value in the same transaction that writes the value. Stamping in a separate later pass invites rows that briefly exist without provenance.

def stamp_record(db, parcel_id: str, zoning_code: str, token: str) -> None:
    db.execute(
        """
        UPDATE parcels
           SET zoning_code = %s, lineage_id = %s
         WHERE parcel_id = %s
        """,
        (zoning_code, token, parcel_id),
    )

Step 5 — Persist the append-only audit row

Model the row with pydantic so it is validated and frozen, then append it to a store that rejects updates and deletes. The row is the record that survives even after the operational value is later corrected.

from datetime import datetime, timezone
from pydantic import BaseModel, Field


class LineageRow(BaseModel):
    lineage_id: str
    parcel_id: str
    target_value: str
    source_address: str
    transform_id: str
    mapping_version: str
    recorded_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
    model_config = {"frozen": True}


def append_audit(sink, row: LineageRow) -> None:
    """sink must INSERT only; UPDATE/DELETE grants are revoked from this role."""
    sink.insert(row.model_dump(mode="json"))


append_audit(audit_sink, LineageRow(
    lineage_id=token, parcel_id="048-217-009", target_value="TR",
    source_address=src_addr, transform_id=transform_id,
    mapping_version=mapping_version,
))

Step 6 — Verify reproducibility

The proof only holds if re-running the chain from the same inputs yields the same token. Fetch the source bytes back from the manifest by their address, recompute, and assert equality.

def verify(manifest, token: str, source_address: str,
           transform_id: str, mapping_version: str) -> bool:
    raw = manifest.get(source_address)                 # exact original bytes
    recomputed = lineage_hash(content_address(raw), transform_id, mapping_version)
    return recomputed == token


assert verify(manifest, token, src_addr, transform_id, mapping_version)

Verification and testing

Confirm the stamp is genuine evidence rather than a decorative column:

  • Reproduction is bit-for-bit. Re-running Steps 1–3 against the archived source bytes must reproduce an identical lin: token. Any drift means a non-deterministic input — usually re-serialized floats or a local-timezone timestamp folded into canonicalization.
  • The source refetches. manifest.get(src_addr) must return the exact bytes whose hash equals src_addr. A mismatch means the manifest stored extracted text or a re-encoded copy instead of the original.
  • Every new row is stamped. Assert SELECT count(*) FROM parcels WHERE lineage_id IS NULL trends to zero for records written after the cutover, and alert if it does not.
  • The audit store rejects mutation. Attempt an UPDATE against the audit table with the application role and confirm it is denied. A log you can edit is not an audit log.
  • Backfill does not overwrite. When reprocessing history, confirm new lineage rows are appended with the new transform_id; existing rows for the same parcel must remain untouched.

Failure recovery

When stamping fails mid-run, the pipeline must fail closed and stay reproducible:

  • Quarantine, never store bare. If any of the three inputs is missing — an empty mapping_version, an unresolved transform version — do not write the value with a null lineage_id. Route the record to a quarantine partition with the reason, and fix the input rather than degrading the guarantee.
  • Source bytes lost. If the original PDF was never archived, you cannot honestly compute a source address after the fact. Record the row as provenance: irrecoverable rather than hashing whatever the portal serves today, which would assert a binding you cannot defend.
  • Audit append failed after the value wrote. Because Step 4 and Step 5 can straddle a crash, reconcile on restart: any parcels row whose lineage_id has no matching audit row is re-stamped from the manifest, and any orphaned manifest object without a value is left in place for the next run.
  • Backfilling the historical gap. For rows written before the cutover, reprocess from the archived source if it exists; where it does not, mark them irrecoverable and report the count, so the coverage gap is itself an auditable fact rather than a silent unknown.

Frequently asked questions

Can I add a lineage_id to existing rows without the original source?

Only honestly if you archived the source bytes. Computing a source address from whatever the portal serves today asserts that the value came from those bytes, which is exactly the claim you cannot prove for a historical row. If the original is gone, mark the row provenance-irrecoverable and report it; a truthful gap is worth more than a fabricated hash.

Why chain the inputs with a \x00 delimiter instead of just concatenating?

Plain concatenation lets different input splits collide — (“ab”, “c”) and (“a”, “bc”) produce the same string. A NUL byte cannot appear inside a content address, transform id, or version string, so joining on it guarantees the three fields are unambiguously separated and two distinct derivations can never hash to the same token.

Does the lineage hash change when I fix a typo in the zoning value?

Not by itself — the token chains the source, transform, and mapping, not the output value. If you correct a value, you should be correcting one of those inputs (a new transform version, say), which does change the token, and you append a fresh audit row. Editing only the stored value without a corresponding input change is a red flag the audit trail is designed to surface.

How does this differ from a database row version or updated_at column?

A row version tells you the record changed and when; it says nothing about which municipal publication or mapping produced the value. Lineage binds the value to its external origin, so it survives migrations, restores, and refactors that would reset row versions. The two are complementary — keep both — but only lineage answers an auditor’s reconstruction request.