Enforce quality gates on output publication

docs/QUALITY_POLICY.md

@@ -1,10 +1,10 @@
-# Datasource Stage Quality Policy
+# Datasource Quality Policy
 
 Datasource quality policy is a deterministic JSON contract stored in
-`governance.quality_policy`. A tabular stage inherits the current target
+`governance.quality_policy`. A tabular stage or producer publication inherits
-Datasource policy unless the stage has its own governed definition. Validation
+the current target Datasource policy unless it supplies its own governed
-runs before the stage is stored, but a failed stage remains available for
+definition. Stage validation runs before the stage is stored, but a failed
-inspection and correction through a new stage.
+stage remains available for inspection and correction through a new stage.
 
 ## Contract
 
@@ -78,6 +78,18 @@ contain only affected counts and at most 25 one-based row numbers; they never
 copy field values. Promotion copies this validation object into the immutable
 materialization provenance and records the policy hash in the audit event.
 
+Producer publication uses the same gate before any catalogue target,
+materialization, or publication record is persisted. A rejected output has no
+partial catalogue effect. Successful output materializations retain the exact
+validation result, policy version/hash, and schema classification in
+`provenance.publication_validation`; the publication record retains the same
+evidence for operational inspection. Dataflow, Workflow, and Reporting can
+therefore consume an immutable output reference without re-running a possibly
+changed quality policy. Publication also emits a transactional
+`datasource.publication.published` audit/platform event; an enabled Audit
+module stores it in the durable outbox, while reduced installations deliver it
+through Core after commit.
+
 Approval authority, approval expiry, and retention/deletion execution remain
 separate work under `govoplan-datasources#2`. Until those contracts are added,
 no JSON flag is treated as an approval and no stage is deleted automatically.

src/govoplan_datasources/backend/manifest.py

@@ -406,15 +406,16 @@ manifest = ModuleManifest(
         ),
         DocumentationTopic(
             id="datasources.quality-gates",
-            title="Validate a Datasource stage",
+            title="Validate staged and produced Datasource revisions",
-            summary="Apply deterministic quality rules and schema-change policy before staged data can become consumable.",
+            summary="Apply deterministic quality rules and schema-change policy before staged or produced data can become consumable.",
             body=(
                 "Tabular stages evaluate configured row count, required-field, field-shape, nullability, uniqueness, numeric-range, and bounded "
                 "referential-set rules. Errors keep the stage inspectable but block promotion; warnings stay visible and permit an explicit "
                 "promotion. Updates compare the detected schema with the current target and classify each change as compatible, warning, or "
                 "breaking. Diagnostics expose counts and bounded row numbers, never field values. The policy version and hash, diagnostics, and "
-                "schema diff are copied into immutable materialization provenance when promotion succeeds. Approval and retention execution are "
+                "schema diff are copied into immutable materialization provenance when promotion succeeds. Producer publication applies the same "
-                "not inferred from arbitrary JSON flags and remain separate governed lifecycle work."
+                "gate before any catalogue effect, retains validation evidence on the immutable output revision, and emits a transactional terminal event. Approval and retention execution "
+                "are not inferred from arbitrary JSON flags and remain separate governed lifecycle work."
             ),
             layer="available",
             documentation_types=("admin", "user"),

src/govoplan_datasources/backend/service.py

@@ -10,6 +10,7 @@ from typing import Any, cast
 from sqlalchemy import exists, func, or_, select
 from sqlalchemy.orm import Session
 
+from govoplan_core.audit.logging import audit_event
 from govoplan_core.auth import ApiPrincipal, has_scope
 from govoplan_core.core.datasources import (
     DatasourceAccessError,
@@ -108,12 +109,37 @@ class SqlDatasourceProvider:
             return existing
 
         actor_id = _actor_id(api_principal)
+        target, governance, baseline_schema = _publication_validation_context(
+            db,
+            tenant_id=api_principal.tenant_id,
+            request=request,
+        )
+        validation = validate_stage(
+            rows=prepared.rows,
+            schema=prepared.schema,
+            quality_policy=governance.quality_policy,
+            baseline_schema=baseline_schema,
+        )
+        if validation["valid"] is not True:
+            errors = validation.get("errors", [])
+            error_codes = ", ".join(
+                str(item.get("code") or "validation.error")
+                for item in errors
+                if isinstance(item, Mapping)
+            )
+            suffix = f" ({error_codes})" if error_codes else ""
+            raise DatasourceValidationError(
+                "Published output failed governed quality or schema validation"
+                f"{suffix}."
+            )
         datasource = _publication_target(
             db,
             tenant_id=api_principal.tenant_id,
             actor_id=actor_id,
             request=request,
             prepared=prepared,
+            target=target,
+            governance=governance,
         )
         materialization = _append_materialization(
             db,
@@ -126,7 +152,10 @@ class SqlDatasourceProvider:
             frozen=request.freeze,
             frozen_label=request.frozen_label,
             source_timestamp=request.source_timestamp,
-            provenance=_publication_provenance(request, prepared),
+            provenance={
+                **_publication_provenance(request, prepared),
+                "publication_validation": validation,
+            },
             metadata=dict(request.metadata),
             set_current=request.set_current,
         )
@@ -139,6 +168,28 @@ class SqlDatasourceProvider:
             request=request,
             prepared=prepared,
         )
+        audit_event(
+            db,
+            tenant_id=api_principal.tenant_id,
+            user_id=getattr(api_principal.user, "id", None)
+            or api_principal.account_id,
+            api_key_id=api_principal.api_key_id,
+            action="datasource.publication.published",
+            object_type="datasource_publication",
+            object_id=publication.id,
+            details={
+                "producer_module": prepared.producer_module,
+                "producer_run_ref": prepared.producer_run_ref,
+                "datasource_ref": _datasource_ref(datasource.id),
+                "materialization_ref": _materialization_ref(materialization.id),
+                "fingerprint": prepared.fingerprint,
+                "row_count": len(prepared.rows),
+                "policy_hash": validation["policy_hash"],
+                "schema_classification": validation["schema_change"][
+                    "classification"
+                ],
+            },
+        )
         return DatasourcePublicationResult(
             ref=_publication_ref(publication.id),
             status=publication.status,
@@ -1607,19 +1658,13 @@ def _publication_target(
     actor_id: str | None,
     request: DatasourcePublicationRequest,
     prepared: _PreparedPublication,
+    target: DatasourceRecord | None,
+    governance: DatasourceGovernance,
 ) -> DatasourceRecord:
-    if request.target_datasource_ref:
+    if target is not None:
-        datasource = _required_datasource(
+        datasource = target
-            session,
-            tenant_id=tenant_id,
-            datasource_ref=request.target_datasource_ref,
-        )
-        if datasource.mode == "live" or datasource.shape != "tabular":
-            raise DatasourceValidationError(
-                "Produced rows require a static or cached tabular datasource."
-            )
         if request.governance is not None:
-            _apply_datasource_governance(datasource, request.governance)
+            _apply_datasource_governance(datasource, governance)
             datasource.updated_by = actor_id
         return datasource
     name = str(request.name or "").strip()
@@ -1665,14 +1710,47 @@ def _publication_target(
     )
     _apply_datasource_governance(
         datasource,
-        request.governance
+        governance,
-        or _default_governance(mode="static", provider_ref=None),
     )
     session.add(datasource)
     session.flush()
     return datasource
 
 
+def _publication_validation_context(
+    session: Session,
+    *,
+    tenant_id: str,
+    request: DatasourcePublicationRequest,
+) -> tuple[
+    DatasourceRecord | None,
+    DatasourceGovernance,
+    tuple[DatasourceField, ...] | None,
+]:
+    if not request.target_datasource_ref:
+        return (
+            None,
+            request.governance
+            or _default_governance(mode="static", provider_ref=None),
+            None,
+        )
+    target = _required_datasource(
+        session,
+        tenant_id=tenant_id,
+        datasource_ref=request.target_datasource_ref,
+        for_update=True,
+    )
+    if target.mode == "live" or target.shape != "tabular":
+        raise DatasourceValidationError(
+            "Produced rows require a static or cached tabular datasource."
+        )
+    return (
+        target,
+        request.governance or _datasource_governance(target),
+        _fields(target.schema_) if target.current_materialization_id else None,
+    )
+
+
 def _publication_provenance(
     request: DatasourcePublicationRequest,
     prepared: _PreparedPublication,
@@ -1710,6 +1788,10 @@ def _create_publication_record(
             "row_count": len(prepared.rows),
             "set_current": request.set_current,
             "frozen": request.freeze,
+            "validation": materialization.provenance_.get(
+                "publication_validation",
+                {},
+            ),
         },
         created_by=actor_id,
     )

tests/test_lifecycle.py

@@ -7,6 +7,7 @@ from sqlalchemy.orm import sessionmaker
 
 from govoplan_core.auth import ApiPrincipal
 from govoplan_core.core.access import PrincipalRef
+from govoplan_core.core.change_sequence import ChangeSequenceEntry
 from govoplan_core.core.datasources import (
     CAPABILITY_DATASOURCE_ORIGINS,
     DatasourceAccessError,
@@ -154,6 +155,7 @@ class DatasourceLifecycleTests(unittest.TestCase):
                 DatasourceMaterializationRecord.__table__,
                 DatasourceStageRecord.__table__,
                 DatasourcePublicationRecord.__table__,
+                ChangeSequenceEntry.__table__,
             ],
         )
         self.Session = sessionmaker(bind=self.engine)
@@ -175,6 +177,7 @@ class DatasourceLifecycleTests(unittest.TestCase):
                 DatasourcePayloadRecord.__table__,
                 DatasourceGovernanceReferenceRecord.__table__,
                 DatasourceRecord.__table__,
+                ChangeSequenceEntry.__table__,
             ],
         )
         self.engine.dispose()
@@ -651,6 +654,106 @@ class DatasourceLifecycleTests(unittest.TestCase):
                 ),
             )
 
+    def test_publication_enforces_quality_before_persisting_any_effect(self) -> None:
+        producer = principal(scopes=(SOURCE_WRITE_SCOPE,))
+        blocked = DatasourcePublicationRequest(
+            producer_module="dataflow",
+            producer_run_ref="dataflow-run:blocked",
+            idempotency_key="blocked-output",
+            name="Blocked output",
+            source_name="blocked_output",
+            rows=({"id": 1},),
+            governance=DatasourceGovernance(
+                quality_policy={
+                    "version": "minimum-two-v1",
+                    "rules": [
+                        {
+                            "id": "minimum-two",
+                            "type": "row_count",
+                            "minimum": 2,
+                        }
+                    ],
+                },
+            ),
+        )
+
+        with self.assertRaisesRegex(
+            DatasourceValidationError,
+            r"quality or schema validation \(quality\.row_count\)",
+        ):
+            self.provider.publish_rows(self.session, producer, request=blocked)
+
+        self.assertEqual(0, self.session.query(DatasourceRecord).count())
+        self.assertEqual(
+            0,
+            self.session.query(DatasourceMaterializationRecord).count(),
+        )
+        self.assertEqual(
+            0,
+            self.session.query(DatasourcePublicationRecord).count(),
+        )
+
+    def test_publication_preserves_quality_and_schema_evidence(self) -> None:
+        producer = principal(scopes=(SOURCE_WRITE_SCOPE,))
+        first = self.provider.publish_rows(
+            self.session,
+            producer,
+            request=DatasourcePublicationRequest(
+                producer_module="dataflow",
+                producer_run_ref="dataflow-run:first",
+                idempotency_key="governed-output-first",
+                name="Governed output",
+                source_name="governed_output",
+                rows=({"id": 1},),
+                governance=DatasourceGovernance(
+                    quality_policy={
+                        "version": "governed-output-v1",
+                        "rules": [
+                            {
+                                "id": "unique-id",
+                                "type": "unique",
+                                "fields": ["id"],
+                            },
+                        ],
+                        "schema_policy": {"field_added_required": "warning"},
+                    },
+                ),
+            ),
+        )
+        second = self.provider.publish_rows(
+            self.session,
+            producer,
+            request=DatasourcePublicationRequest(
+                producer_module="dataflow",
+                producer_run_ref="dataflow-run:second",
+                idempotency_key="governed-output-second",
+                target_datasource_ref=first.datasource.ref,
+                rows=({"id": 2, "result": "match"},),
+            ),
+        )
+
+        validation = second.materialization.provenance["publication_validation"]
+        self.assertTrue(validation["valid"])
+        self.assertEqual("governed-output-v1", validation["policy_version"])
+        self.assertEqual(
+            "warning",
+            validation["schema_change"]["classification"],
+        )
+        self.assertEqual(
+            "schema.field_added_required",
+            validation["warnings"][0]["code"],
+        )
+        record = self.session.get(
+            DatasourcePublicationRecord,
+            second.ref.removeprefix("publication:"),
+        )
+        self.assertIsNotNone(record)
+        assert record is not None
+        self.assertEqual(
+            validation["policy_hash"],
+            record.details_["validation"]["policy_hash"],
+        )
+
     def test_publication_requires_source_write_scope(self) -> None:
         with self.assertRaises(DatasourceAccessError):
             self.provider.publish_rows(