Replaying Pipeline Transformations for Audit

When an auditor asks how a smudged handwritten line became a clean ledger entry, the honest answer is a replay, not a paragraph. This page shows how to record each transformation a record undergoes as an ordered, hash-chained event, so the whole journey from raw log line to normalized record can be re-executed deterministically and verified byte for byte. It extends Data Lineage & Provenance Tracking from “where did this value come from” to “what exactly was done to it, in what order” — the reconstruction that FAA 14 CFR § 43.11 and the electronic-records guidance in AC 43-9C ultimately depend on.

Prerequisites & inputs

Replay requires that every transformation be pure and versioned. A transform takes an input state and produces an output state with no hidden dependence on wall-clock time, random seeds, or mutable global config; anything variable must be captured as an explicit parameter in the event. You also need the field-level origin data from Capturing Field-Level Provenance in Parsed MRO Records, because replay begins from the raw captured bytes and re-derives everything after them.

The running example is a defect rectification on N512AA, ATA 36-11 (pneumatic bleed air), component VZ-8820-3, whose raw log line reads RECT bleed leak clamp rplcd P/N VZ8820-3 @ 21044.2 FH.

Deterministic transform replay

Deterministic transform replayRawlineTransformeventsHashchainReplayengineVerifiedoutput

Step-by-step implementation

Step 1 — record each transform as an event. An event captures the transform name, its version, the SHA-256 of the input state, the SHA-256 of the output state, and any parameters. The input hash of event n+1 must equal the output hash of event n, forming a chain that is broken by any gap or reordering.

Step 2 — chain the events cryptographically. Each event also stores a link_hash derived from the previous event’s link_hash plus its own content. A tampered or missing event anywhere in the sequence changes every subsequent link, so verification is a single pass.

Step 3 — replay by re-executing. Given the raw input and the ordered transforms, re-run each function and assert that the recomputed output hash matches the recorded one. A divergence localizes the exact transform whose behavior drifted.

from __future__ import annotations

import hashlib
import json
import logging
from typing import Callable, Dict, List

from pydantic import BaseModel, ConfigDict, Field

logging.basicConfig(level=logging.INFO, format="%(asctime)s | %(levelname)s | %(message)s")
logger = logging.getLogger("mro.lineage.replay")


def _sha256(text: str) -> str:
    return hashlib.sha256(text.encode("utf-8")).hexdigest()


class TransformEvent(BaseModel):
    """One ordered, hash-chained transformation.

    Supports reconstruction of the record per FAA 14 CFR § 43.11, which
    governs the content of records for inspections and required maintenance.
    """

    model_config = ConfigDict(frozen=True)

    seq: int = Field(ge=0)
    name: str = Field(min_length=1)
    version: str = Field(min_length=1)
    input_hash: str = Field(pattern=r"^[0-9a-f]{64}$")
    output_hash: str = Field(pattern=r"^[0-9a-f]{64}$")
    params: Dict[str, str] = Field(default_factory=dict)
    link_hash: str = Field(pattern=r"^[0-9a-f]{64}$")


def link(prev_link: str, seq: int, name: str, out_hash: str) -> str:
    return _sha256(f"{prev_link}|{seq}|{name}|{out_hash}")


def build_chain(raw: str, transforms: List[tuple[str, str, Callable[[str], str]]]) -> \
        tuple[str, List[TransformEvent]]:
    state = raw
    prev_link = _sha256("GENESIS")
    events: List[TransformEvent] = []
    for seq, (name, version, fn) in enumerate(transforms):
        in_hash = _sha256(state)
        state = fn(state)
        out_hash = _sha256(state)
        prev_link = link(prev_link, seq, name, out_hash)
        events.append(TransformEvent(seq=seq, name=name, version=version,
                                     input_hash=in_hash, output_hash=out_hash,
                                     link_hash=prev_link))
    return state, events


def replay(raw: str, transforms: List[tuple[str, str, Callable[[str], str]]],
           events: List[TransformEvent]) -> bool:
    state = raw
    prev_link = _sha256("GENESIS")
    for seq, (name, version, fn) in enumerate(transforms):
        ev = events[seq]
        if _sha256(state) != ev.input_hash:
            logger.error("input divergence at seq=%d name=%s", seq, name)
            return False
        state = fn(state)
        prev_link = link(prev_link, seq, name, _sha256(state))
        if _sha256(state) != ev.output_hash or prev_link != ev.link_hash:
            logger.error("output/link divergence at seq=%d name=%s", seq, name)
            return False
    logger.info("replay verified over %d transforms", len(events))
    return True


if __name__ == "__main__":
    pipeline: List[tuple[str, str, Callable[[str], str]]] = [
        ("strip_prefix", "1.0.0", lambda s: s.replace("RECT ", "", 1)),
        ("normalize_pn", "2.1.0", lambda s: s.replace("VZ8820-3", "VZ-8820-3")),
        ("to_json", "1.2.0", lambda s: json.dumps({"raw": s}, sort_keys=True)),
    ]
    final, chain = build_chain("RECT bleed leak clamp rplcd P/N VZ8820-3 @ 21044.2 FH", pipeline)
    print("verified:", replay("RECT bleed leak clamp rplcd P/N VZ8820-3 @ 21044.2 FH",
                              pipeline, chain))

Worked example

Running the N512AA line through the three-transform pipeline produces a chain of three events. The normalize_pn transform rewrites VZ8820-3 to the canonical VZ-8820-3; its event records an input hash matching the previous output, so the chain is continuous. replay re-executes all three and confirms every output and link hash:

2026-06-03 11:47:08 | INFO | replay verified over 3 transforms
verified: True

Now suppose an engineer later “fixes” normalize_pn to also uppercase the string. Re-running replay against the stored chain fails precisely at seq=1:

2026-06-03 11:52:41 | ERROR | output/link divergence at seq=1 name=normalize_pn

The auditor learns not just that the record changed, but that the second transform is where behavior diverged — and because the link hashes chain forward, no later event could have hidden it. Records that pass replay can move on to the contract checks in Schema Validation & Error Handling with their history proven intact.

Edge cases & compliance gotchas

Non-deterministic transforms. A step that folds in datetime.now() or a dictionary iteration order will replay differently and raise false divergences. Capture every such input as an explicit params entry, and sort keys in any JSON serialization so output is byte-stable.

Unicode normalization mid-chain. If one transform emits NFC and a later replay environment defaults to NFD, hashes over technician names like Lefèvre diverge for no substantive reason. Pin a normalization form at the chain’s start and treat it as part of the transform contract.

Floating-point drift on flight hours. Re-parsing 21044.2 through a different locale or float formatter can yield 21044.20, changing the output hash. Serialize numeric fields through a fixed formatter, not the platform default.

Transform version renames. Renaming a function without bumping its version silently invalidates old chains. Treat the name/version pair as an immutable identity and register a new transform rather than mutating an existing one. Held to these rules, the event chain turns an audit from an argument into an experiment: the auditor re-runs history and watches it reproduce, or watches it fail at the exact step that broke.