Pin reporting datasets to published Dataflow runs

docs/ADMIN_GUIDE.md

@@ -27,6 +27,10 @@ published by a source-owning module. Do not expose another module's ORM or an
 unbounded SQL connection as a report source. Configure an explicit schema,
 freshness policy, source fingerprint expectations, purpose, privacy,
 retention, and a row-policy provider where source access alone is not enough.
+For a Dataflow source, `source_run_ref` optionally pins one successful run that
+published an immutable Datasource materialization. Reporting passes this pin
+through rather than re-executing the pipeline. The current principal must still
+be authorized for the Dataflow definition and the exact Datasource output.
 
 On PostgreSQL installations, Reporting compiles bounded semantic filters,
 grouping, measures, calculated measures, ordering, offsets, and limits into a

docs/USER_GUIDE.md

@@ -28,6 +28,11 @@ the exact definition and output. Warnings explain freshness, inferred schema,
 or provider diagnostics. A failed quality gate records a failed execution and
 does not publish a result.
 
+A report configured against an exact published Dataflow run reads that run's
+immutable Datasource materialization. It does not rerun the flow with current
+inputs. The evidence identifies the Dataflow run and materialization, and access
+to both is checked again when the report runs.
+
 The **Effective access** explanation states when dimensions, measures, source
 rows, or actions were removed by Policy. A result with no hidden elements says
 so explicitly; catalogue visibility never grants access to protected detail.

src/govoplan_reporting/backend/execution.py

@@ -501,6 +501,7 @@ def _read_dataset(
             request=DataflowDatasetRequest(
                 pipeline_ref=dataset.source_ref,
                 revision=dataset.source_revision or 0,
+                run_ref=dataset.source_run_ref,
                 parameters=source_parameters,
                 row_limit=2_000,
                 expected_definition_hash=dataset.definition_hash,

src/govoplan_reporting/backend/manifest.py

@@ -488,7 +488,7 @@ manifest = ModuleManifest(
                 "schedules, exports, and publication providers replace unchecked SQL in the "
                 "presentation layer. PostgreSQL executes bounded semantic plans when available. "
                 "Signed drill contexts reauthorize contributor rows, and Files/Mail publication "
-                "adapters retain idempotent evidence. Dataflow and module read models remain source owners."
+                "adapters retain idempotent evidence. A dataset may pin one successful published Dataflow run, which is read from its exact Datasource materialization after both source boundaries reauthorize the current principal. Dataflow and module read models remain source owners."
             ),
             layer="available",
             documentation_types=("admin", "user"),

src/govoplan_reporting/backend/schemas.py

@@ -101,6 +101,7 @@ class DatasetDefinition(BaseModel):
     source_kind: Literal["dataflow", "read_model", "static"]
     source_ref: str = Field(min_length=1, max_length=500)
     source_revision: int | None = Field(default=None, ge=1)
+    source_run_ref: str | None = Field(default=None, min_length=1, max_length=500)
     definition_hash: str | None = Field(default=None, min_length=1, max_length=128)
     source_parameters: dict[str, Any] = Field(default_factory=dict)
     static_rows: list[dict[str, Any]] = Field(default_factory=list, max_length=2_000)
@@ -130,6 +131,8 @@ class DatasetDefinition(BaseModel):
     def validate_source_pin(self) -> "DatasetDefinition":
         if self.source_kind == "dataflow" and self.source_revision is None:
             raise ValueError("Dataflow datasets require a pinned source revision.")
+        if self.source_run_ref is not None and self.source_kind != "dataflow":
+            raise ValueError("Only Dataflow datasets can pin a source run.")
         if self.source_kind == "static" and not self.static_rows:
             raise ValueError("Static analytical datasets require static_rows.")
         names = [item.name for item in self.fields]

tests/test_reporting_service.py

@@ -8,6 +8,10 @@ from sqlalchemy import create_engine
 from sqlalchemy.orm import Session
 
 from govoplan_core.db.base import Base
+from govoplan_core.core.dataflows import (
+    CAPABILITY_DATAFLOW_DATASET_OUTPUT,
+    DataflowDatasetResult,
+)
 from govoplan_core.core.files import (
     CAPABILITY_FILES_ARTIFACT_STORE,
     ManagedArtifactRef,
@@ -132,6 +136,34 @@ class ArtifactStore:
         )
 
 
+class DataflowOutput:
+    def __init__(self, rows: list[dict[str, object]]) -> None:
+        self.rows = tuple(dict(item) for item in rows)
+        self.last_request = None
+
+    def list_outputs(self, *_args, **_kwargs):
+        return ()
+
+    def read_output(self, _session, _principal, *, request):
+        self.last_request = request
+        return DataflowDatasetResult(
+            pipeline_ref=request.pipeline_ref,
+            revision=request.revision,
+            definition_hash=request.expected_definition_hash or "pipeline-hash",
+            rows=self.rows,
+            total_rows=len(self.rows),
+            truncated=False,
+            output_hash="d" * 64,
+            executor_version="duckdb-v1",
+            run_ref=request.run_ref,
+            source_fingerprints=(
+                {"node_id": "source", "fingerprint": "source-v1"},
+            ),
+            generated_at=NOW,
+            provenance={"immutable_run": bool(request.run_ref)},
+        )
+
+
 class ReportingServiceTests(unittest.TestCase):
     def setUp(self) -> None:
         self.engine = create_engine("sqlite+pysqlite:///:memory:")
@@ -352,6 +384,44 @@ class ReportingServiceTests(unittest.TestCase):
             )
         self.assertTrue(raised.exception.execution_id)
 
+    def test_dataflow_dataset_can_pin_an_exact_published_run(self) -> None:
+        payload = dataset_payload()
+        rows = list(payload.pop("static_rows"))
+        payload.update(
+            {
+                "source_kind": "dataflow",
+                "source_ref": "pipeline:monthly-comparison",
+                "source_revision": 4,
+                "source_run_ref": "dataflow-run:published-july",
+                "definition_hash": "pipeline-hash",
+            }
+        )
+        self._create("dataset", "dataset-1", payload)
+        self._create("semantic_model", "semantic-1", semantic_payload())
+        self._create("report", "report-1", report_payload())
+        provider = DataflowOutput(rows)
+        registry = CapabilityRegistry()
+        registry.providers[CAPABILITY_DATAFLOW_DATASET_OUTPUT] = provider
+
+        result = execute_report(
+            self.session,
+            self.principal,
+            registry=registry,
+            report_id="report-1",
+            report_revision=1,
+            parameters={},
+            query=None,
+            idempotency_key="published-dataflow-run",
+        )
+
+        self.assertEqual("succeeded", result["status"])
+        self.assertIsNotNone(provider.last_request)
+        self.assertEqual(
+            "dataflow-run:published-july",
+            provider.last_request.run_ref,
+        )
+        self.assertTrue(result["provenance"]["source"]["immutable_run"])
+
     def test_restricted_access_and_service_scope_guards(self) -> None:
         self._create_report_graph(
             report_access={