47.2 Data Lineage and Provenance Standards

1. The W3C PROV Data Model

Suppose a regulator asks you to prove, right now, that the model behind a clinical decision was never trained on retracted patient data: could your lab answer in under five minutes? The W3C PROV family of specifications (Moreau & Missier, 2013) exists so the answer can be yes. PROV defines a domain-agnostic, formally specified data model for provenance. The model supports multiple serialization formats: JSON, Resource Description Framework (RDF), Extensible Markup Language (XML), and a human-readable notation called PROV-N (a compact textual syntax where each statement maps directly to one PROV relation). The model is built on three core types and the relationships between them. Figure 47.2.1 illustrates W3C PROV data model applied to an ML pipeline.

Data provenance is a structured record of the origin, transformations, and ownership chain of every artifact in a computational pipeline. Without provenance, reproducing a result means reconstructing every decision, code version, and data snapshot by hand. That reconstruction becomes impractical as pipelines grow in complexity or span multiple teams. Each time a process consumes inputs and produces outputs, a provenance system records that relationship as a typed graph edge. The edge links the input artifacts, the transformation activity, and the output artifacts. Use formal provenance standards (such as W3C PROV or OpenLineage, covered below) whenever your work crosses organizational boundaries, faces external audits, or requires long-term reproducibility. For single-researcher exploratory work, lightweight experiment loggers like MLflow tracking may suffice. In short: provenance turns "I think this model used clean data" into "I can prove it in one query."

• Entity: a physical, digital, or conceptual thing. In our context: a dataset, a model checkpoint, a configuration file, or an evaluation report.
• Activity: something that occurs over a period of time and acts upon entities. In our context: a preprocessing run, a training job, or an evaluation pass.
• Agent: something that bears responsibility for an activity. In our context: a researcher, an automated pipeline, or a Discovery Workbench agent.

Figure 47.9 illustrates how these three core types interconnect. Each arrow represents one of the six PROV relations listed in Table 47.3, showing the bipartite structure (where a bipartite graph is one whose nodes split into two disjoint sets, here entities and activities, with edges running only between sets) that links entities to the activities that consumed or produced them, and the agents responsible for those activities.

These three types are connected by a small set of relations (summarized in Table 47.3) that capture the essential provenance semantics:

The wasDerivedFrom relation is a convenience shortcut: it asserts that one entity was produced from another without specifying the intermediate activity. For full provenance, the decomposition into used + wasGeneratedBy through an explicit activity is preferred, because it captures the transformation that connected input to output. Formally, if entity \(e_2\) was derived from entity \(e_1\), there exists an activity \(a\) such that:

2. Implementing PROV in Python

The following example (Figure 47.10) encodes a two-stage ML pipeline (preprocessing followed by training) as a PROV document. The prov Python library provides a complete implementation of the W3C PROV data model with serialization to PROV-N, PROV-JSON, and RDF:

"""Encoding ML pipeline provenance with the W3C PROV data model."""
import prov.model as prov
from prov.dot import prov_to_dot
from datetime import datetime, timezone


def create_pipeline_provenance() -> prov.ProvDocument:
    """Create a PROV document describing a two-stage ML pipeline."""
    doc = prov.ProvDocument()

    # Define namespaces for our identifiers
    doc.set_default_namespace("https://discoveryai.example.org/prov/")
    doc.add_namespace("dvc", "https://discoveryai.example.org/dvc/")
    doc.add_namespace("mlflow", "https://discoveryai.example.org/mlflow/")
    doc.add_namespace("git", "https://discoveryai.example.org/git/")

    # --- Entities (artifacts) ---

    # Raw dataset
    raw_data = doc.entity("dvc:raw_crystals_v2", {
        "prov:type": "dvc:Dataset",
        "dvc:hash": "a1b2c3d4e5f6...",
        "dvc:size_bytes": 1073741824,
        "prov:label": "Raw crystal structure dataset v2",
    })

    # Preprocessing configuration
    preprocess_config = doc.entity("git:config/preprocess.yaml@abc123", {
        "prov:type": "git:ConfigFile",
        "git:commit": "abc123def456",
        "prov:label": "Preprocessing configuration",
    })

    # Processed features (output of preprocessing)
    processed_data = doc.entity("dvc:processed_features_v3", {
        "prov:type": "dvc:Dataset",
        "dvc:hash": "f6e5d4c3b2a1...",
        "dvc:size_bytes": 536870912,
        "prov:label": "Processed feature matrix v3",
    })

    # Training configuration
    train_config = doc.entity("git:config/train.yaml@abc123", {
        "prov:type": "git:ConfigFile",
        "git:commit": "abc123def456",
        "prov:label": "Training configuration (lr=0.0005, layers=6)",
    })

    # Trained model
    model = doc.entity("mlflow:CrystalPredictor/v12", {
        "prov:type": "mlflow:RegisteredModel",
        "mlflow:run_id": "run_20260415_042",
        "mlflow:version": 12,
        "prov:label": "CrystalPredictor model v12",
    })

    # Evaluation metrics
    metrics = doc.entity("mlflow:run_20260415_042/metrics", {
        "prov:type": "mlflow:Metrics",
        "mlflow:mae": 0.042,
        "mlflow:r2": 0.94,
        "prov:label": "Evaluation metrics (MAE=0.042, R2=0.94)",
    })

    # --- Activities (runs) ---

    preprocess_run = doc.activity(
        "dvc:preprocess_run_001",
        startTime=datetime(2026, 4, 15, 10, 0, tzinfo=timezone.utc),
        endTime=datetime(2026, 4, 15, 10, 5, tzinfo=timezone.utc),
        other_attributes={
            "prov:type": "dvc:PipelineStage",
            "prov:label": "Data preprocessing",
            "git:commit": "abc123def456",
        }
    )

    train_run = doc.activity(
        "mlflow:train_run_042",
        startTime=datetime(2026, 4, 15, 10, 10, tzinfo=timezone.utc),
        endTime=datetime(2026, 4, 15, 12, 30, tzinfo=timezone.utc),
        other_attributes={
            "prov:type": "mlflow:TrainingRun",
            "prov:label": "Model training",
            "git:commit": "abc123def456",
            "mlflow:experiment_id": "crystal_property_prediction",
        }
    )

    # --- Agents ---

    researcher = doc.agent("git:alice@lab.org", {
        "prov:type": "prov:Person",
        "prov:label": "Alice (lead researcher)",
    })

    pipeline = doc.agent("dvc:automated_pipeline", {
        "prov:type": "prov:SoftwareAgent",
        "prov:label": "DVC automated pipeline",
    })

    # --- Relations ---

    # Preprocessing stage
    doc.wasGeneratedBy(processed_data, preprocess_run)
    doc.used(preprocess_run, raw_data)
    doc.used(preprocess_run, preprocess_config)
    doc.wasAssociatedWith(preprocess_run, pipeline)

    # Training stage
    doc.wasGeneratedBy(model, train_run)
    doc.wasGeneratedBy(metrics, train_run)
    doc.used(train_run, processed_data)
    doc.used(train_run, train_config)
    doc.wasAssociatedWith(train_run, pipeline)

    # Delegation: pipeline acts on behalf of researcher
    doc.actedOnBehalfOf(pipeline, researcher)

    # Derivation shortcuts (redundant but useful for querying)
    doc.wasDerivedFrom(processed_data, raw_data)
    doc.wasDerivedFrom(model, processed_data)

    # Attribution
    doc.wasAttributedTo(model, researcher)

    return doc


# Create and serialize the provenance document
doc = create_pipeline_provenance()

# Serialize to PROV-JSON (machine-readable)
prov_json = doc.serialize(format="json")
print("PROV-JSON document created")

# Serialize to PROV-N (human-readable)
prov_n = doc.serialize(format="provn")
print("\nPROV-N representation:")
print(prov_n[:500])

# Visualize as a graph (requires graphviz)
dot = prov_to_dot(doc)
dot.write_png("provenance_graph.jpg")
print("\nProvenance graph saved to provenance_graph.jpg")

The PROV-N serialization of this document, shown in Figure 47.11, is a human-readable representation that reads almost like natural language:

document
  prefix dvc  <https://discoveryai.example.org/dvc/>
  prefix mlflow <https://discoveryai.example.org/mlflow/>

  entity(dvc:raw_crystals_v2, [prov:type="dvc:Dataset"])
  entity(dvc:processed_features_v3, [prov:type="dvc:Dataset"])
  entity(mlflow:CrystalPredictor/v12, [prov:type="mlflow:RegisteredModel"])

  activity(dvc:preprocess_run_001, 2026-04-15T10:00:00Z, 2026-04-15T10:05:00Z)
  activity(mlflow:train_run_042, 2026-04-15T10:10:00Z, 2026-04-15T12:30:00Z)

  wasGeneratedBy(dvc:processed_features_v3, dvc:preprocess_run_001)
  used(dvc:preprocess_run_001, dvc:raw_crystals_v2)
  wasGeneratedBy(mlflow:CrystalPredictor/v12, mlflow:train_run_042)
  used(mlflow:train_run_042, dvc:processed_features_v3)

  wasDerivedFrom(mlflow:CrystalPredictor/v12, dvc:raw_crystals_v2)
endDocument

3. The OpenLineage Specification

While W3C PROV provides a general-purpose provenance model, OpenLineage is a specification designed specifically for runtime lineage collection in data pipelines. Where PROV captures provenance as a document authored after the fact, OpenLineage captures lineage as a stream of events emitted during pipeline execution. This event-driven model integrates naturally with orchestration tools (Airflow, Dagster, Spark) that already emit lifecycle events. (As of 2024, OpenLineage has reached version 1.0 and added column-level lineage support, with additional integrations for Apache Flink and Great Expectations.)

OpenLineage defines three event types that map to pipeline execution phases:

• RunEvent (START): emitted when a job begins execution. Contains the job name, namespace, and input datasets.
• RunEvent (RUNNING): optionally emitted during execution with progress information.
• RunEvent (COMPLETE/FAIL/ABORT): emitted when a job finishes. Contains output datasets, metrics, and error information (for failures).

Each event carries facets, where a facet is a named, schema-validated metadata extension that describes a specific aspect of the job, dataset, or run. Standard facets include SQL queries, schema information, data quality metrics, and source code locations. Custom facets allow domain-specific metadata (crystal structure properties, simulation parameters, assay conditions). The code in Figure 47.12 demonstrates how to emit these events from Python.

"""OpenLineage event emission for ML pipeline lineage tracking."""
import json
import uuid
from datetime import datetime, timezone
from dataclasses import dataclass, field, asdict
from typing import Optional


@dataclass
class OpenLineageDataset:
    """An OpenLineage dataset reference."""
    namespace: str        # e.g., "s3://my-bucket"
    name: str             # e.g., "data/crystals.parquet"
    facets: dict = field(default_factory=dict)


@dataclass
class OpenLineageJob:
    """An OpenLineage job reference."""
    namespace: str        # e.g., "https://discoveryai.example.org"
    name: str             # e.g., "crystal_preprocessing"
    facets: dict = field(default_factory=dict)


@dataclass
class OpenLineageRunEvent:
    """A single OpenLineage event capturing pipeline state."""
    eventType: str        # "START", "RUNNING", "COMPLETE", "FAIL", "ABORT"
    eventTime: str        # ISO 8601
    run: dict             # {"runId": uuid}
    job: dict             # Job reference
    inputs: list = field(default_factory=list)
    outputs: list = field(default_factory=list)
    producer: str = "https://discoveryai.example.org/pipeline/v1"


class OpenLineageEmitter:
    """Emit OpenLineage events during pipeline execution."""

    def __init__(self, job_namespace: str, job_name: str):
        self.run_id = str(uuid.uuid4())
        self.job = OpenLineageJob(namespace=job_namespace, name=job_name)
        self.events: list[dict] = []

    def emit_start(
        self,
        inputs: list[OpenLineageDataset],
    ) -> dict:
        """Emit a START event when the job begins."""
        event = OpenLineageRunEvent(
            eventType="START",
            eventTime=datetime.now(timezone.utc).isoformat(),
            run={"runId": self.run_id},
            job=asdict(self.job),
            inputs=[asdict(ds) for ds in inputs],
        )
        event_dict = asdict(event)
        self.events.append(event_dict)
        self._send(event_dict)
        return event_dict

    def emit_complete(
        self,
        outputs: list[OpenLineageDataset],
        run_facets: Optional[dict] = None,
    ) -> dict:
        """Emit a COMPLETE event when the job finishes successfully."""
        event = OpenLineageRunEvent(
            eventType="COMPLETE",
            eventTime=datetime.now(timezone.utc).isoformat(),
            run={
                "runId": self.run_id,
                "facets": run_facets or {},
            },
            job=asdict(self.job),
            outputs=[asdict(ds) for ds in outputs],
        )
        event_dict = asdict(event)
        self.events.append(event_dict)
        self._send(event_dict)
        return event_dict

    def emit_fail(self, error_message: str) -> dict:
        """Emit a FAIL event when the job encounters an error."""
        event = OpenLineageRunEvent(
            eventType="FAIL",
            eventTime=datetime.now(timezone.utc).isoformat(),
            run={
                "runId": self.run_id,
                "facets": {
                    "errorMessage": {
                        "_producer": "https://discoveryai.example.org",
                        "_schemaURL": "https://openlineage.io/spec/facets/1-0-0/ErrorMessageRunFacet.json",
                        "message": error_message,
                        "programmingLanguage": "python",
                    }
                },
            },
            job=asdict(self.job),
        )
        event_dict = asdict(event)
        self.events.append(event_dict)
        self._send(event_dict)
        return event_dict

    def _send(self, event: dict):
        """Send event to the lineage backend (console for demo)."""
        print(f"[OpenLineage] {event['eventType']}: "
              f"{self.job.name} (run {self.run_id[:8]}...)")


# Example: instrument a preprocessing step
emitter = OpenLineageEmitter(
    job_namespace="https://discoveryai.example.org",
    job_name="crystal_feature_extraction",
)

# Emit START with input datasets
emitter.emit_start(inputs=[
    OpenLineageDataset(
        namespace="s3://crystal-data",
        name="raw/structures_v2.parquet",
        facets={
            "schema": {
                "fields": [
                    {"name": "formula", "type": "string"},
                    {"name": "space_group", "type": "integer"},
                    {"name": "lattice_params", "type": "array"},
                ]
            },
            "dataSource": {"uri": "s3://crystal-data/raw/structures_v2.parquet"},
        },
    )
])

# ... perform preprocessing ...

# Emit COMPLETE with output datasets
emitter.emit_complete(
    outputs=[
        OpenLineageDataset(
            namespace="s3://crystal-data",
            name="processed/features_v3.parquet",
            facets={
                "schema": {
                    "fields": [
                        {"name": "feature_vector", "type": "array"},
                        {"name": "target_bandgap", "type": "float"},
                    ]
                },
                "rowCount": {"rowCount": 47523},
            },
        )
    ],
    run_facets={
        "processing": {
            "durationMs": 305000,
            "recordsProcessed": 47523,
            "recordsDropped": 142,
        }
    },
)

4. PROV vs. OpenLineage: Complementary Standards

W3C PROV and OpenLineage serve different but complementary roles. Table 47.4 summarizes the key differences. Understanding when to use each prevents over-engineering:

The practical pattern is: emit OpenLineage events during pipeline execution, aggregate them in a lineage backend (such as Marquez), and export the aggregated lineage as W3C PROV documents for publication, archival, or regulatory submission. Figure 47.13 shows the bridge code that converts between the two:

"""Convert OpenLineage events to W3C PROV documents."""
import prov.model as prov
from datetime import datetime


def openlineage_to_prov(events: list[dict]) -> prov.ProvDocument:
    """Convert a sequence of OpenLineage events into a PROV document.

    Groups events by run ID, creates PROV Activities for each run,
    PROV Entities for each dataset, and connects them with
    used/wasGeneratedBy relations.
    """
    doc = prov.ProvDocument()
    doc.set_default_namespace("https://discoveryai.example.org/prov/")
    doc.add_namespace("ol", "https://openlineage.io/")

    # Group events by run ID
    runs: dict[str, list[dict]] = {}
    for event in events:
        run_id = event["run"]["runId"]
        runs.setdefault(run_id, []).append(event)

    seen_entities = set()

    for run_id, run_events in runs.items():
        # Find the START and terminal events
        start_event = next(
            (e for e in run_events if e["eventType"] == "START"), None
        )
        end_event = next(
            (e for e in run_events
             if e["eventType"] in ("COMPLETE", "FAIL", "ABORT")),
            None,
        )

        if not start_event:
            continue

        job_name = start_event["job"]["name"]

        # Create the PROV Activity
        start_time = datetime.fromisoformat(start_event["eventTime"])
        end_time = (
            datetime.fromisoformat(end_event["eventTime"])
            if end_event else None
        )

        activity_id = f"ol:run/{run_id}"
        doc.activity(
            activity_id,
            startTime=start_time,
            endTime=end_time,
            other_attributes={
                "prov:type": "ol:JobRun",
                "prov:label": job_name,
                "ol:runId": run_id,
            },
        )

        # Create entities for input datasets and link with 'used'
        for inp in start_event.get("inputs", []):
            entity_id = f"ol:dataset/{inp['namespace']}/{inp['name']}"
            if entity_id not in seen_entities:
                doc.entity(entity_id, {
                    "prov:type": "ol:Dataset",
                    "prov:label": inp["name"],
                    "ol:namespace": inp["namespace"],
                })
                seen_entities.add(entity_id)
            doc.used(activity_id, entity_id)

        # Create entities for output datasets and link with
        # 'wasGeneratedBy'
        if end_event and end_event["eventType"] == "COMPLETE":
            for out in end_event.get("outputs", []):
                entity_id = f"ol:dataset/{out['namespace']}/{out['name']}"
                if entity_id not in seen_entities:
                    doc.entity(entity_id, {
                        "prov:type": "ol:Dataset",
                        "prov:label": out["name"],
                        "ol:namespace": out["namespace"],
                    })
                    seen_entities.add(entity_id)
                doc.wasGeneratedBy(entity_id, activity_id)

    return doc


# Convert collected OpenLineage events to PROV
prov_doc = openlineage_to_prov(emitter.events)
print(prov_doc.serialize(format="provn"))

5. Run Graphs and Provenance Queries

With provenance captured in a structured format, we can build a run graph: a queryable graph database of all experiments, artifacts, and their relationships. The implementation below uses NetworkX (a Python library for creating, manipulating, and analyzing graph data structures) to store the graph in memory. The run graph enables questions that are impossible to answer with flat experiment logs:

• What data was used to train the model currently deployed in production? (backward lineage)
• Which downstream models are affected if I retract this dataset? (forward lineage / impact analysis)
• Show me all runs that used dataset version X with learning rate > 0.001. (filtered search)
• What changed between the model that scored 0.94 and the one that scored 0.91? (diff analysis)

"""Queryable run graph using NetworkX for provenance analysis."""
import networkx as nx
from dataclasses import dataclass


class RunGraph:
    """A queryable graph of experiment runs and artifacts."""

    def __init__(self):
        self.graph = nx.DiGraph()

    def add_entity(self, entity_id: str, **attrs):
        """Add an artifact (entity) node to the graph."""
        self.graph.add_node(entity_id, node_type="entity", **attrs)

    def add_activity(self, activity_id: str, **attrs):
        """Add a run (activity) node to the graph."""
        self.graph.add_node(activity_id, node_type="activity", **attrs)

    def add_used(self, activity_id: str, entity_id: str):
        """Record that an activity used an entity as input."""
        self.graph.add_edge(entity_id, activity_id, relation="used")

    def add_generated(self, entity_id: str, activity_id: str):
        """Record that an activity generated an entity as output."""
        self.graph.add_edge(activity_id, entity_id, relation="wasGeneratedBy")

    def backward_lineage(self, entity_id: str) -> set[str]:
        """Find all ancestors of an entity (what was it derived from?)."""
        return nx.ancestors(self.graph, entity_id)

    def forward_impact(self, entity_id: str) -> set[str]:
        """Find all descendants of an entity (what depends on it?)."""
        return nx.descendants(self.graph, entity_id)

    def find_runs_using(
        self,
        entity_id: str,
        filter_fn=None,
    ) -> list[str]:
        """Find all runs that consumed a specific entity."""
        runs = []
        for successor in self.graph.successors(entity_id):
            node = self.graph.nodes[successor]
            if node.get("node_type") == "activity":
                if filter_fn is None or filter_fn(node):
                    runs.append(successor)
        return runs

    def diff_runs(self, run_a: str, run_b: str) -> dict:
        """Compare two runs: what inputs, params, and outputs differ?"""
        inputs_a = {
            n for n in self.graph.predecessors(run_a)
            if self.graph.nodes[n].get("node_type") == "entity"
        }
        inputs_b = {
            n for n in self.graph.predecessors(run_b)
            if self.graph.nodes[n].get("node_type") == "entity"
        }

        outputs_a = {
            n for n in self.graph.successors(run_a)
            if self.graph.nodes[n].get("node_type") == "entity"
        }
        outputs_b = {
            n for n in self.graph.successors(run_b)
            if self.graph.nodes[n].get("node_type") == "entity"
        }

        return {
            "inputs_only_in_a": inputs_a - inputs_b,
            "inputs_only_in_b": inputs_b - inputs_a,
            "shared_inputs": inputs_a & inputs_b,
            "outputs_only_in_a": outputs_a - outputs_b,
            "outputs_only_in_b": outputs_b - outputs_a,
            "params_a": self.graph.nodes[run_a],
            "params_b": self.graph.nodes[run_b],
        }

    def regeneration_order(self, entity_id: str) -> list[str]:
        """Topological order (a linear ordering of nodes such that every
        dependency appears before the node that depends on it) of
        activities needed to regenerate an entity."""
        ancestors = self.backward_lineage(entity_id)
        # Filter to activity nodes only
        activities = {
            n for n in ancestors
            if self.graph.nodes[n].get("node_type") == "activity"
        }
        # Add any activity that directly produces the target
        for pred in self.graph.predecessors(entity_id):
            if self.graph.nodes[pred].get("node_type") == "activity":
                activities.add(pred)
        # Return in topological (execution) order
        subgraph = self.graph.subgraph(ancestors | {entity_id})
        topo = list(nx.topological_sort(subgraph))
        return [n for n in topo if n in activities]


# Build a run graph
rg = RunGraph()

# Add entities
rg.add_entity("raw_data_v2", label="Raw crystals", dvc_hash="a1b2c3...")
rg.add_entity("features_v3", label="Processed features", dvc_hash="d4e5f6...")
rg.add_entity("model_v12", label="Trained model", mlflow_version=12)
rg.add_entity("config_v1", label="Training config", lr=0.0005)

# Add activities
rg.add_activity("preprocess_001", label="Preprocessing", duration_s=305)
rg.add_activity("train_042", label="Training", duration_s=8400, lr=0.0005)

# Add relations
rg.add_used("preprocess_001", "raw_data_v2")
rg.add_generated("features_v3", "preprocess_001")
rg.add_used("train_042", "features_v3")
rg.add_used("train_042", "config_v1")
rg.add_generated("model_v12", "train_042")

# Query: what does model_v12 depend on?
lineage = rg.backward_lineage("model_v12")
print(f"Model v12 lineage ({len(lineage)} ancestors):")
for ancestor in lineage:
    node = rg.graph.nodes[ancestor]
    print(f"  {ancestor}: {node.get('label', 'unknown')}")

# Query: if raw_data_v2 is retracted, what is affected?
impact = rg.forward_impact("raw_data_v2")
print(f"\nImpact of retracting raw_data_v2: {len(impact)} downstream artifacts")

# Query: regeneration plan for model_v12
plan = rg.regeneration_order("model_v12")
print(f"\nRegeneration plan for model_v12: {plan}")

6. Provenance for the Discovery Workbench

The run graph gives us a general-purpose engine for lineage queries, but a production research platform needs provenance recording woven into its pipeline execution, not bolted on after the fact.

The Discovery Workbench introduced in Chapter 6 includes a provenance layer that stores lineage for all Workbench operations. The standards from this section connect directly to the Workbench architecture through a provenance recorder that instruments Workbench pipeline stages. Figure 47.15 shows the recorder implementation.

"""Discovery Workbench provenance integration."""
import prov.model as prov
from datetime import datetime, timezone
from contextlib import contextmanager
from typing import Generator


class WorkbenchProvenanceRecorder:
    """Records provenance for Discovery Workbench pipeline stages.

    Integrates with the Workbench's event system to automatically
    capture lineage as experiments execute.
    """

    def __init__(self, workbench_id: str):
        self.workbench_id = workbench_id
        self.doc = prov.ProvDocument()
        self.doc.set_default_namespace(
            f"https://discoveryai.example.org/workbench/{workbench_id}/"
        )
        self.doc.add_namespace("wb", "https://discoveryai.example.org/workbench/")

        # Register the Workbench as an agent
        self.doc.agent(f"wb:{workbench_id}", {
            "prov:type": "wb:DiscoveryWorkbench",
            "prov:label": f"Discovery Workbench ({workbench_id})",
        })

    @contextmanager
    def track_stage(
        self,
        stage_name: str,
        input_ids: list[str],
    ) -> Generator[dict, None, None]:
        """Context manager that tracks a pipeline stage's provenance.

        Usage:
            with recorder.track_stage("preprocess", ["raw_data"]) as ctx:
                result = preprocess(raw_data)
                ctx["outputs"] = [("processed_data", result_hash)]
                ctx["metrics"] = {"records": 47523}
        """
        stage_id = f"stage:{stage_name}:{datetime.now(timezone.utc).strftime('%Y%m%d%H%M%S')}"
        start_time = datetime.now(timezone.utc)

        # Create the activity
        self.doc.activity(stage_id, startTime=start_time, other_attributes={
            "prov:type": "wb:PipelineStage",
            "prov:label": stage_name,
        })

        # Link to the Workbench agent
        self.doc.wasAssociatedWith(stage_id, f"wb:{self.workbench_id}")

        # Record input usage
        for input_id in input_ids:
            if not self.doc.get_record(input_id):
                self.doc.entity(input_id, {"prov:type": "wb:Artifact"})
            self.doc.used(stage_id, input_id)

        # Yield context for the stage to populate outputs
        context = {"outputs": [], "metrics": {}}
        try:
            yield context
        finally:
            end_time = datetime.now(timezone.utc)

            # Update the activity with end time
            # (prov library handles this via the activity record)

            # Record outputs
            for output_id, output_hash in context.get("outputs", []):
                self.doc.entity(output_id, {
                    "prov:type": "wb:Artifact",
                    "wb:contentHash": output_hash,
                })
                self.doc.wasGeneratedBy(output_id, stage_id)

                # Derivation from each input
                for input_id in input_ids:
                    self.doc.wasDerivedFrom(output_id, input_id)

    def export_prov_json(self) -> str:
        """Export the full provenance document as PROV-JSON."""
        return self.doc.serialize(format="json")

    def export_prov_n(self) -> str:
        """Export the full provenance document as human-readable PROV-N."""
        return self.doc.serialize(format="provn")


# Example: Workbench pipeline with automatic provenance
recorder = WorkbenchProvenanceRecorder("wb_crystal_2026")

# Stage 1: preprocessing
with recorder.track_stage("feature_extraction", ["raw_crystals_v2"]) as ctx:
    # ... actual preprocessing would happen here ...
    ctx["outputs"] = [("features_v3", "d4e5f6a7b8...")]
    ctx["metrics"] = {"records_in": 47665, "records_out": 47523}

# Stage 2: training
with recorder.track_stage("model_training", ["features_v3", "config_v1"]) as ctx:
    # ... actual training would happen here ...
    ctx["outputs"] = [("model_v12", "c3b2a1f6e5...")]
    ctx["metrics"] = {"epochs": 87, "val_mae": 0.042}

# Export provenance for archival
print(recorder.export_prov_n())

7. Provenance Integrity and Tamper Detection

Recording provenance automatically, as the Workbench recorder does, solves the collection problem, but it raises a new question: how can anyone verify that the recorded history has not been altered after the fact?

Provenance is only useful when trustworthy. A record that can be altered after the fact is worse than none: it creates false confidence. Two mechanisms strengthen integrity: signing provenance documents (using cryptographic signatures to bind a provenance record to the identity of its author, covered conceptually here but not implemented until Section 47.4) and chaining records. Figure 47.16 demonstrates the chaining approach:

"""Tamper-evident provenance chains using hash linking."""
import hashlib
import json
from datetime import datetime, timezone


class ProvenanceChain:
    """An append-only, tamper-evident chain of provenance records.

    Each record includes the hash of the previous record, forming
    a hash chain (similar to a blockchain's structure, but without
    consensus or mining). Modifying any historical record breaks
    the chain, making tampering detectable.
    """

    def __init__(self):
        self.records: list[dict] = []

    def _hash_record(self, record: dict) -> str:
        """Compute the hash of a provenance record."""
        canonical = json.dumps(record, sort_keys=True).encode("utf-8")
        return hashlib.sha256(canonical).hexdigest()

    def append(self, record: dict) -> dict:
        """Append a new record to the chain."""
        # Link to the previous record
        if self.records:
            record["_prev_hash"] = self._hash_record(self.records[-1])
        else:
            record["_prev_hash"] = "GENESIS"

        record["_timestamp"] = datetime.now(timezone.utc).isoformat()
        record["_index"] = len(self.records)
        record["_hash"] = self._hash_record(record)

        self.records.append(record)
        return record

    def verify_integrity(self) -> tuple[bool, list[int]]:
        """Verify the entire chain. Returns (is_valid, broken_indices)."""
        broken = []

        for i, record in enumerate(self.records):
            # Verify self-hash
            expected_hash = record.pop("_hash")
            actual_hash = self._hash_record(record)
            record["_hash"] = expected_hash

            if actual_hash != expected_hash:
                broken.append(i)
                continue

            # Verify chain linkage
            if i > 0:
                prev_record = self.records[i - 1].copy()
                expected_prev = self._hash_record(prev_record)
                if record["_prev_hash"] != expected_prev:
                    broken.append(i)

        return len(broken) == 0, broken


# Build a provenance chain
chain = ProvenanceChain()

chain.append({
    "event": "dataset_registered",
    "entity": "raw_crystals_v2",
    "hash": "a1b2c3d4...",
    "agent": "alice@lab.org",
})

chain.append({
    "event": "preprocessing_complete",
    "activity": "preprocess_001",
    "inputs": ["raw_crystals_v2"],
    "outputs": ["features_v3"],
    "agent": "automated_pipeline",
})

chain.append({
    "event": "model_trained",
    "activity": "train_042",
    "inputs": ["features_v3", "config_v1"],
    "outputs": ["model_v12"],
    "metrics": {"mae": 0.042},
    "agent": "automated_pipeline",
})

# Verify the chain
is_valid, broken = chain.verify_integrity()
print(f"Chain integrity: {'VALID' if is_valid else 'BROKEN'}")
print(f"Records: {len(chain.records)}, Broken: {broken}")

8. The Provenance Maturity Model

Hash chains and signed documents represent the most rigorous end of the provenance spectrum, but most teams do not need that level of assurance from day one; the right investment depends on the stakes of the project.

Not every project needs full W3C PROV compliance with tamper-evident chains. A practical approach is to adopt provenance incrementally, advancing through the maturity levels shown in Table 47.5 as the project's needs grow:

The recipe in Section 47.3 implements Level 3 (full artifact DAG with hash verification) and includes optional Level 4 extensions for teams that need auditability.