Serve exact published runs as datasets

README.md

@@ -152,6 +152,14 @@ logical row disappeared. It never silently applies a correction to business
 data; a downstream governed transform or Workflow handoff must interpret the
 recorded action.
 
+Reporting consumers may either evaluate a pinned pipeline revision or pin one
+successful published run. An exact run pin is immutable: it cannot be supplied
+new parameters, and Dataflow reads only the recorded Datasource materialization
+through the provider-neutral catalogue capability. Both Dataflow run authority
+and Datasource row access are rechecked for the current principal; the returned
+lineage retains the run, publication, datasource, materialization, fingerprint,
+and governance snapshot.
+
 ## Development
 
 ```bash

src/govoplan_dataflow/backend/dataset_output.py

@@ -11,6 +11,11 @@ from govoplan_core.core.dataflows import (
     DataflowRunConflictError,
     DataflowRunUnavailableError,
 )
+from govoplan_core.core.datasources import (
+    DatasourceError,
+    DatasourceReadRequest,
+    datasource_catalogue,
+)
 from govoplan_core.security.time import utc_now
 from govoplan_dataflow.backend.backends.base import ExecutionBudget
 from govoplan_dataflow.backend.executor import PipelineExecutionError
@@ -20,6 +25,7 @@ from govoplan_dataflow.backend.service import (
     _execute_pipeline_preview,
     get_pipeline,
     get_pipeline_revision,
+    get_pipeline_run,
     list_pipelines,
 )
 from govoplan_dataflow.backend.subflows import substitute_parameters
@@ -110,6 +116,17 @@ class SqlDataflowDatasetOutputProvider:
             raise DataflowRunConflictError(
                 "The pinned Dataflow definition hash no longer matches the requested revision."
             )
+        if request.run_ref:
+            return _read_published_run_output(
+                typed_session,
+                typed_principal,
+                registry=self.registry,
+                request=request,
+                pipeline=pipeline,
+                revision=revision,
+                policy_decision=decision.to_dict(),
+                row_limit=row_limit,
+            )
         graph = PipelineGraph.model_validate(
             substitute_parameters(revision.graph, dict(request.parameters))
         )
@@ -179,6 +196,143 @@ def dataset_output_provider(context: object | None = None):
     return SqlDataflowDatasetOutputProvider(getattr(context, "registry", None))
 
 
+def _read_published_run_output(
+    session,
+    principal: ApiPrincipal,
+    *,
+    registry: object | None,
+    request: DataflowDatasetRequest,
+    pipeline,
+    revision,
+    policy_decision: dict[str, object],
+    row_limit: int,
+) -> DataflowDatasetResult:
+    if request.parameters:
+        raise DataflowRunConflictError(
+            "An immutable published run cannot be evaluated with new parameters."
+        )
+    run = get_pipeline_run(
+        session,
+        tenant_id=principal.tenant_id,
+        run_ref=request.run_ref or "",
+    )
+    if run.pipeline_id != pipeline.id or run.pipeline_revision_id != revision.id:
+        raise DataflowRunConflictError(
+            "The published run does not belong to the pinned Dataflow revision."
+        )
+    if run.definition_hash != revision.content_hash:
+        raise DataflowRunConflictError(
+            "The published run definition evidence does not match the pinned revision."
+        )
+    if run.status != "succeeded":
+        raise DataflowRunUnavailableError(
+            "Only a successful Dataflow run can be used as an immutable dataset."
+        )
+    if not run.output_datasource_ref or not run.output_materialization_ref:
+        raise DataflowRunUnavailableError(
+            "The successful Dataflow run has no immutable Datasource publication."
+        )
+    provider = datasource_catalogue(registry)
+    if provider is None:
+        raise DataflowRunUnavailableError(
+            "The Datasource catalogue required by this published run is not enabled."
+        )
+
+    rows: list[dict[str, object]] = []
+    materialization = None
+    total_rows = 0
+    try:
+        while len(rows) < row_limit:
+            remaining = row_limit - len(rows)
+            page = provider.read_datasource(
+                session,
+                principal,
+                request=DatasourceReadRequest(
+                    datasource_ref=run.output_datasource_ref,
+                    materialization_ref=run.output_materialization_ref,
+                    limit=min(500, remaining),
+                    offset=len(rows),
+                    expected_fingerprint=(
+                        materialization.fingerprint
+                        if materialization is not None
+                        else None
+                    ),
+                ),
+            )
+            if (
+                page.materialization is None
+                or page.materialization.ref != run.output_materialization_ref
+            ):
+                raise DataflowRunConflictError(
+                    "The Datasource provider returned a different output materialization."
+                )
+            if len(page.rows) > remaining:
+                raise DataflowRunConflictError(
+                    "The Datasource provider exceeded the requested output window."
+                )
+            materialization = page.materialization
+            total_rows = page.total_rows
+            rows.extend(dict(item) for item in page.rows)
+            if not page.truncated:
+                break
+            if not page.rows:
+                raise DataflowRunUnavailableError(
+                    "The Datasource provider made no progress while reading the published output."
+                )
+    except DatasourceError as exc:
+        raise DataflowRunUnavailableError(
+            "The immutable Datasource output is unavailable to the current principal."
+        ) from exc
+
+    source_fingerprints = tuple(dict(item) for item in run.source_fingerprints)
+    if request.expected_source_fingerprints and not _fingerprints_match(
+        request.expected_source_fingerprints,
+        source_fingerprints,
+    ):
+        raise DataflowRunConflictError(
+            "Dataflow source fingerprints differ from the pinned run evidence."
+        )
+    output_hash = hashlib.sha256(
+        json.dumps(
+            rows,
+            sort_keys=True,
+            separators=(",", ":"),
+            ensure_ascii=True,
+            default=str,
+        ).encode("utf-8")
+    ).hexdigest()
+    if materialization is None:
+        raise DataflowRunUnavailableError(
+            "The Datasource provider returned no materialization evidence."
+        )
+    return DataflowDatasetResult(
+        pipeline_ref=pipeline.id,
+        revision=revision.revision,
+        definition_hash=revision.content_hash,
+        rows=tuple(rows),
+        total_rows=total_rows,
+        truncated=len(rows) < total_rows,
+        output_hash=output_hash,
+        executor_version=run.executor_version,
+        run_ref=f"dataflow-run:{run.id}",
+        source_fingerprints=source_fingerprints,
+        diagnostics=tuple(dict(item) for item in run.diagnostics),
+        generated_at=run.finished_at or materialization.created_at,
+        provenance={
+            "module": "dataflow",
+            "scope_type": pipeline.scope_type,
+            "scope_id": pipeline.scope_id,
+            "policy_decision": policy_decision,
+            "immutable_run": True,
+            "publication_ref": run.output_publication_ref,
+            "datasource_ref": run.output_datasource_ref,
+            "materialization_ref": run.output_materialization_ref,
+            "materialization_fingerprint": materialization.fingerprint,
+            "governance": materialization.governance.to_dict(),
+        },
+    )
+
+
 def _fingerprints_match(expected, actual) -> bool:
     def normalized(values):
         return sorted(

src/govoplan_dataflow/backend/manifest.py

@@ -270,6 +270,7 @@ DOCUMENTATION = (
             "states whether work stopped, completed, failed, or requires operator reconciliation. Scheduled, event, "
             "and queued execution is partitioned by tenant module entitlement before a run is claimed. Disabling "
             "Dataflow stops new admission and leaves accepted runs available for an explicit operator decision."
+            " Reporting may pin a successful published run; Dataflow then rechecks run authority and Datasource access and reads only the exact recorded materialization without reparameterizing or re-executing it."
         ),
         layer="available",
         documentation_types=("admin", "user"),

tests/test_dataset_output.py

@@ -1,14 +1,25 @@
 from __future__ import annotations
 
+from datetime import UTC, datetime
 import unittest
 
 from govoplan_core.auth import ApiPrincipal
 from govoplan_core.core.access import PrincipalRef
 from govoplan_core.core.dataflows import DataflowDatasetRequest, DataflowRunConflictError
+from govoplan_core.core.datasources import (
+    CAPABILITY_DATASOURCE_CATALOGUE,
+    DatasourceDescriptor,
+    DatasourceMaterialization,
+    DatasourceReadResult,
+)
 from govoplan_core.db.base import Base
 from govoplan_core.db.session import configure_database, reset_database
 from govoplan_dataflow.backend.dataset_output import SqlDataflowDatasetOutputProvider
-from govoplan_dataflow.backend.db.models import DataflowPipeline, DataflowPipelineRevision
+from govoplan_dataflow.backend.db.models import (
+    DataflowPipeline,
+    DataflowPipelineRevision,
+    DataflowRun,
+)
 from govoplan_dataflow.backend.schemas import PipelineGraph
 from govoplan_dataflow.backend.service import definition_hash
 
@@ -26,12 +37,71 @@ def principal(tenant_id: str = "tenant-1") -> ApiPrincipal:
     )
 
 
+class PublishedOutputCatalogue:
+    rows = (
+        {"recipient_key": "published", "email": "published@example.test"},
+    )
+
+    def list_datasources(self, *_args, **_kwargs):
+        return ()
+
+    def get_datasource(self, *_args, **_kwargs):
+        return None
+
+    def list_materializations(self, *_args, **_kwargs):
+        return ()
+
+    def read_datasource(self, _session, _principal, *, request):
+        materialization = DatasourceMaterialization(
+            ref="materialization:published-1",
+            datasource_ref="datasource:published-1",
+            revision=1,
+            state="published",
+            fingerprint="f" * 64,
+            row_count=len(self.rows),
+            created_at=datetime(2026, 8, 4, 9, 0, tzinfo=UTC),
+        )
+        rows = self.rows[request.offset : request.offset + request.limit]
+        return DatasourceReadResult(
+            datasource=DatasourceDescriptor(
+                ref="datasource:published-1",
+                source_name="published_output",
+                name="Published output",
+                kind="custom",
+                mode="static",
+                shape="tabular",
+                fingerprint=materialization.fingerprint,
+                current_materialization_ref=materialization.ref,
+                row_count=len(self.rows),
+            ),
+            materialization=materialization,
+            rows=rows,
+            total_rows=len(self.rows),
+            truncated=request.offset + len(rows) < len(self.rows),
+        )
+
+
+class CapabilityRegistry:
+    def __init__(self, catalogue) -> None:
+        self.catalogue = catalogue
+
+    def has_capability(self, name: str) -> bool:
+        return name == CAPABILITY_DATASOURCE_CATALOGUE
+
+    def capability(self, name: str):
+        return self.catalogue if self.has_capability(name) else None
+
+
 class DataflowDatasetOutputTests(unittest.TestCase):
     def setUp(self) -> None:
         self.database = configure_database("sqlite:///:memory:")
         Base.metadata.create_all(
             self.database.engine,
-            tables=[DataflowPipeline.__table__, DataflowPipelineRevision.__table__],
+            tables=[
+                DataflowPipeline.__table__,
+                DataflowPipelineRevision.__table__,
+                DataflowRun.__table__,
+            ],
         )
 
     def tearDown(self) -> None:
@@ -131,6 +201,67 @@ class DataflowDatasetOutputTests(unittest.TestCase):
                         expected_definition_hash="wrong",
                     ),
                 )
+
+            run = DataflowRun(
+                id="run-published-1",
+                tenant_id="tenant-1",
+                pipeline_id=pipeline.id,
+                pipeline_revision_id=revision.id,
+                run_type="published",
+                status="succeeded",
+                execution_backend="duckdb",
+                environment="production",
+                executor_version="duckdb-v1",
+                definition_hash=content_hash,
+                request_={},
+                source_fingerprints=[
+                    {"node_id": "source", "fingerprint": "source-v1"}
+                ],
+                result_schema=[],
+                diagnostics=[],
+                output_row_count=1,
+                output_publication_ref="publication:published-1",
+                output_datasource_ref="datasource:published-1",
+                output_materialization_ref="materialization:published-1",
+                finished_at=datetime(2026, 8, 4, 9, 0, tzinfo=UTC),
+            )
+            session.add(run)
+            session.flush()
+            published_provider = SqlDataflowDatasetOutputProvider(
+                CapabilityRegistry(PublishedOutputCatalogue())
+            )
+            published = published_provider.read_output(
+                session,
+                principal(),
+                request=DataflowDatasetRequest(
+                    pipeline_ref=pipeline.id,
+                    revision=1,
+                    run_ref="dataflow-run:run-published-1",
+                    expected_definition_hash=content_hash,
+                    expected_source_fingerprints=(
+                        {"node_id": "source", "fingerprint": "source-v1"},
+                    ),
+                ),
+            )
+            self.assertEqual("published", published.rows[0]["recipient_key"])
+            self.assertEqual("dataflow-run:run-published-1", published.run_ref)
+            self.assertTrue(published.provenance["immutable_run"])
+            self.assertEqual(
+                "materialization:published-1",
+                published.provenance["materialization_ref"],
+            )
+
+            with self.assertRaises(DataflowRunConflictError):
+                published_provider.read_output(
+                    session,
+                    principal(),
+                    request=DataflowDatasetRequest(
+                        pipeline_ref=pipeline.id,
+                        revision=1,
+                        run_ref="dataflow-run:run-published-1",
+                        parameters={"changed": True},
+                    ),
+                )
             with self.assertRaises(DataflowRunConflictError):
                 provider.read_output(
                     session,