feat(datasources): publish durable artifact outputs

src/govoplan_datasources/backend/service.py

@@ -14,6 +14,7 @@ from govoplan_core.audit.logging import audit_event
 from govoplan_core.auth import ApiPrincipal, has_scope
 from govoplan_core.core.datasources import (
     DatasourceAccessError,
+    DatasourceArtifactReference,
     DatasourceDescriptor,
     DatasourceError,
     DatasourceField,
@@ -25,6 +26,7 @@ from govoplan_core.core.datasources import (
     DatasourceOriginReadRequest,
     DatasourcePublicationRequest,
     DatasourcePublicationResult,
+    DatasourcePublicationStatus,
     DatasourceReadRequest,
     DatasourceReadResult,
     DatasourceStage,
@@ -46,6 +48,7 @@ from govoplan_datasources.backend.payloads import (
     DatasourcePayloadBackend,
     PayloadBackendRegistry,
     create_database_rows_payload,
+    create_external_payload_reference,
     payload_for_materialization,
     validate_payload_size,
 )
@@ -72,9 +75,11 @@ class _PreparedPublication:
     producer_module: str
     producer_run_ref: str
     idempotency_key: str
-    rows: tuple[dict[str, Any], ...]
+    rows: tuple[dict[str, Any], ...] | None
+    artifact: DatasourceArtifactReference | None
     schema: tuple[DatasourceField, ...]
     fingerprint: str
+    row_count: int
     byte_count: int
     request_hash: str
 
@@ -120,9 +125,8 @@ class SqlDatasourceProvider:
             tenant_id=api_principal.tenant_id,
             request=request,
         )
-        validation = validate_stage(
-            rows=prepared.rows,
-            schema=prepared.schema,
+        validation, publication_status = _validate_publication(
+            prepared=prepared,
             quality_policy=governance.quality_policy,
             baseline_schema=baseline_schema,
         )
@@ -138,6 +142,12 @@ class SqlDatasourceProvider:
                 "Published output failed governed quality or schema validation"
                 f"{suffix}."
             )
+        reusable_payload = self._publication_payload(
+            db,
+            tenant_id=api_principal.tenant_id,
+            actor_id=actor_id,
+            prepared=prepared,
+        )
         datasource = _publication_target(
             db,
             tenant_id=api_principal.tenant_id,
@@ -146,11 +156,14 @@ class SqlDatasourceProvider:
             prepared=prepared,
             target=target,
             governance=governance,
+            publish_as_current=(
+                request.set_current and publication_status != "review_required"
+            ),
         )
         materialization = _append_materialization(
             db,
             datasource=datasource,
-            rows=prepared.rows,
+            rows=prepared.rows or (),
             schema=[field_payload(field) for field in prepared.schema],
             fingerprint=prepared.fingerprint,
             byte_count=prepared.byte_count,
@@ -163,7 +176,15 @@ class SqlDatasourceProvider:
                 "publication_validation": validation,
             },
             metadata=dict(request.metadata),
-            set_current=request.set_current,
+            set_current=(
+                request.set_current and publication_status != "review_required"
+            ),
+            reusable_payload=reusable_payload,
+            state=(
+                "review_required"
+                if publication_status == "review_required"
+                else "published"
+            ),
         )
         publication = _create_publication_record(
             db,
@@ -173,6 +194,7 @@ class SqlDatasourceProvider:
             materialization=materialization,
             request=request,
             prepared=prepared,
+            status=publication_status,
         )
         audit_event(
             db,
@@ -180,7 +202,7 @@ class SqlDatasourceProvider:
             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",
+            action=f"datasource.publication.{publication_status}",
             object_type="datasource_publication",
             object_id=publication.id,
             details={
@@ -189,7 +211,12 @@ class SqlDatasourceProvider:
                 "datasource_ref": _datasource_ref(datasource.id),
                 "materialization_ref": _materialization_ref(materialization.id),
                 "fingerprint": prepared.fingerprint,
-                "row_count": len(prepared.rows),
+                "row_count": prepared.row_count,
+                "payload_backend": (
+                    prepared.artifact.backend
+                    if prepared.artifact is not None
+                    else "database_rows"
+                ),
                 "policy_hash": validation["policy_hash"],
                 "schema_classification": validation["schema_change"][
                     "classification"
@@ -198,12 +225,52 @@ class SqlDatasourceProvider:
         )
         return DatasourcePublicationResult(
             ref=_publication_ref(publication.id),
-            status=publication.status,
+            status=cast(DatasourcePublicationStatus, publication.status),
             datasource=_datasource_dto(datasource),
             materialization=_materialization_dto(materialization),
             replayed=False,
         )
 
+    def _publication_payload(
+        self,
+        session: Session,
+        *,
+        tenant_id: str,
+        actor_id: str | None,
+        prepared: _PreparedPublication,
+    ) -> DatasourcePayloadRecord | None:
+        artifact = prepared.artifact
+        if artifact is None:
+            return None
+        backend = self._payload_backends.require(artifact.backend)
+        payload = create_external_payload_reference(
+            session,
+            tenant_id=tenant_id,
+            backend=artifact.backend,
+            locator=artifact.locator,
+            checksum=artifact.checksum,
+            row_count=artifact.row_count,
+            byte_count=artifact.byte_count,
+            actor_id=actor_id,
+            media_type=artifact.media_type,
+            checkpoint=artifact.checkpoint,
+            metadata={
+                **dict(artifact.metadata),
+                "publication_fingerprint": artifact.fingerprint,
+                "artifact_schema": [
+                    field_payload(field) for field in artifact.schema
+                ],
+                "artifact_validation": dict(artifact.validation),
+            },
+        )
+        try:
+            backend.verify(session, payload)
+        except Exception:
+            session.delete(payload)
+            session.flush()
+            raise
+        return payload
+
     def list_datasources(
         self,
         session: object,
@@ -1056,6 +1123,7 @@ def _append_materialization(
     metadata: Mapping[str, object] | None = None,
     set_current: bool,
     reusable_payload: DatasourcePayloadRecord | None = None,
+    state: str = "published",
 ) -> DatasourceMaterializationRecord:
     datasource = _lock_datasource_for_materialization(session, datasource)
     revision = _allocate_materialization_revision(session, datasource)
@@ -1082,7 +1150,7 @@ def _append_materialization(
         tenant_id=datasource.tenant_id,
         datasource_id=datasource.id,
         revision=revision,
-        state="published",
+        state=state,
         schema_version=max(1, int(schema_version or 1)),
         schema_=schema_payload,
         payload_id=payload.id,
@@ -1618,17 +1686,47 @@ def _prepare_publication(
         producer_run_ref=producer_run_ref,
         idempotency_key=idempotency_key,
     )
-    normalized = normalize_rows(request.rows)
-    schema = infer_schema(normalized)
-    fingerprint = fingerprint_rows(normalized, schema)
+    if (request.rows is None) == (request.artifact is None):
+        raise DatasourceValidationError(
+            "A publication requires exactly one inline row payload or durable "
+            "artifact reference."
+        )
+    artifact = request.artifact
+    if artifact is None:
+        normalized: tuple[dict[str, Any], ...] | None = normalize_rows(
+            request.rows or ()
+        )
+        schema = infer_schema(normalized)
+        fingerprint = fingerprint_rows(normalized, schema)
+        row_count = len(normalized)
+        byte_count = encoded_size(normalized)
+    else:
+        normalized = None
+        schema = _validated_artifact_schema(artifact)
+        fingerprint = _validated_sha256(
+            artifact.fingerprint,
+            label="Artifact publication fingerprints",
+        )
+        _validated_sha256(
+            artifact.checksum,
+            label="Artifact publication checksums",
+        )
+        if artifact.row_count < 0 or artifact.byte_count < 0:
+            raise DatasourceValidationError(
+                "Artifact publication sizes cannot be negative."
+            )
+        row_count = artifact.row_count
+        byte_count = artifact.byte_count
     return _PreparedPublication(
         producer_module=producer_module,
         producer_run_ref=producer_run_ref,
         idempotency_key=idempotency_key,
         rows=normalized,
+        artifact=artifact,
         schema=schema,
         fingerprint=fingerprint,
-        byte_count=encoded_size(normalized),
+        row_count=row_count,
+        byte_count=byte_count,
         request_hash=_publication_request_hash(
             request,
             normalized=normalized,
@@ -1637,6 +1735,147 @@ def _prepare_publication(
     )
 
 
+def _validated_artifact_schema(
+    artifact: DatasourceArtifactReference,
+) -> tuple[DatasourceField, ...]:
+    schema = tuple(artifact.schema)
+    names = [field.name.strip() for field in schema]
+    if not schema or any(not name for name in names):
+        raise DatasourceValidationError(
+            "Artifact publications require a non-empty schema."
+        )
+    if len(names) != len(set(names)):
+        raise DatasourceValidationError(
+            "Artifact publication schema field names must be unique."
+        )
+    return schema
+
+
+def _validated_sha256(value: str, *, label: str) -> str:
+    cleaned = value.strip().casefold()
+    try:
+        valid = len(cleaned) == 64 and int(cleaned, 16) >= 0
+    except ValueError:
+        valid = False
+    if not valid:
+        raise DatasourceValidationError(f"{label} must be SHA-256 values.")
+    return cleaned
+
+
+def _validate_publication(
+    *,
+    prepared: _PreparedPublication,
+    quality_policy: Mapping[str, object],
+    baseline_schema: Sequence[DatasourceField] | None,
+) -> tuple[dict[str, object], DatasourcePublicationStatus]:
+    if prepared.artifact is None:
+        validation = validate_stage(
+            rows=prepared.rows or (),
+            schema=prepared.schema,
+            quality_policy=quality_policy,
+            baseline_schema=baseline_schema,
+        )
+        status: DatasourcePublicationStatus = (
+            "published_with_warnings"
+            if validation.get("warnings")
+            else "published"
+        )
+        return validation, status
+
+    raw_rules = quality_policy.get("rules", [])
+    rules = (
+        [item for item in raw_rules if isinstance(item, Mapping)]
+        if isinstance(raw_rules, Sequence)
+        and not isinstance(raw_rules, (str, bytes))
+        else []
+    )
+    metadata_rule_types = {"required_fields", "field"}
+    metadata_policy = {
+        **dict(quality_policy),
+        "rules": [
+            dict(item)
+            for item in rules
+            if str(item.get("type") or "") in metadata_rule_types
+        ],
+    }
+    validation = validate_stage(
+        rows=(),
+        schema=prepared.schema,
+        quality_policy=metadata_policy,
+        baseline_schema=baseline_schema,
+    )
+    if validation["valid"] is not True:
+        return validation, "review_required"
+
+    evidence = dict(prepared.artifact.validation)
+    content_rule_ids = {
+        str(item.get("id") or f"rule-{index + 1}")
+        for index, item in enumerate(rules)
+        if str(item.get("type") or "") not in metadata_rule_types
+    }
+    policy_probe = validate_stage(
+        rows=(),
+        schema=prepared.schema,
+        quality_policy=quality_policy,
+        baseline_schema=baseline_schema,
+    )
+    evidence_rule_ids = {
+        str(item)
+        for item in evidence.get("rules_evaluated", [])
+        if str(item).strip()
+    } if isinstance(evidence.get("rules_evaluated"), Sequence) and not isinstance(
+        evidence.get("rules_evaluated"), (str, bytes)
+    ) else set()
+    evidence_verified = (
+        not content_rule_ids
+        or (
+            evidence.get("valid") is True
+            and str(evidence.get("policy_hash") or "")
+            == str(policy_probe["policy_hash"])
+            and str(evidence.get("payload_checksum") or "").casefold()
+            == prepared.artifact.checksum.casefold()
+            and content_rule_ids.issubset(evidence_rule_ids)
+        )
+    )
+    requested_status = str(evidence.get("status") or "verified")
+    if evidence.get("valid") is False or requested_status == "failed":
+        errors = evidence.get("errors")
+        validation["valid"] = False
+        validation["errors"] = (
+            list(errors)
+            if isinstance(errors, Sequence)
+            and not isinstance(errors, (str, bytes))
+            else [
+                {
+                    "severity": "error",
+                    "code": "quality.artifact_validation",
+                    "message": "Artifact validation evidence reports failure.",
+                }
+            ]
+        )
+        return validation, "review_required"
+
+    evidence_warnings = evidence.get("warnings")
+    if isinstance(evidence_warnings, Sequence) and not isinstance(
+        evidence_warnings, (str, bytes)
+    ):
+        validation["warnings"] = [
+            *list(validation.get("warnings", [])),
+            *list(evidence_warnings),
+        ]
+    validation["artifact_evidence"] = {
+        "verified": evidence_verified,
+        "rules_required": sorted(content_rule_ids),
+        "rules_evaluated": sorted(evidence_rule_ids),
+        "payload_checksum": prepared.artifact.checksum.casefold(),
+    }
+    if not evidence_verified or requested_status == "review_required":
+        return validation, "review_required"
+    if validation.get("warnings") or requested_status == "warning":
+        return validation, "published_with_warnings"
+    return validation, "published"
+
+
 def _validate_publication_identity(
     *,
     producer_module: str,
@@ -1730,7 +1969,7 @@ def _existing_publication_result(
         )
     return DatasourcePublicationResult(
         ref=_publication_ref(publication.id),
-        status=publication.status,
+        status=cast(DatasourcePublicationStatus, publication.status),
         datasource=_datasource_dto(datasource),
         materialization=_materialization_dto(materialization),
         replayed=True,
@@ -1746,6 +1985,7 @@ def _publication_target(
     prepared: _PreparedPublication,
     target: DatasourceRecord | None,
     governance: DatasourceGovernance,
+    publish_as_current: bool,
 ) -> DatasourceRecord:
     if target is not None:
         datasource = target
@@ -1782,9 +2022,9 @@ def _publication_target(
         provider_ref=prepared.producer_run_ref,
         schema_version=1,
         schema_=[field_payload(field) for field in prepared.schema],
-        fingerprint=prepared.fingerprint,
-        row_count=len(prepared.rows),
-        byte_count=prepared.byte_count,
+        fingerprint=prepared.fingerprint if publish_as_current else "",
+        row_count=prepared.row_count if publish_as_current else None,
+        byte_count=prepared.byte_count if publish_as_current else None,
         provenance_={
             **dict(request.provenance),
             "producer_module": prepared.producer_module,
@@ -1859,6 +2099,7 @@ def _create_publication_record(
     materialization: DatasourceMaterializationRecord,
     request: DatasourcePublicationRequest,
     prepared: _PreparedPublication,
+    status: DatasourcePublicationStatus,
 ) -> DatasourcePublicationRecord:
     publication = DatasourcePublicationRecord(
         tenant_id=tenant_id,
@@ -1868,12 +2109,19 @@ def _create_publication_record(
         request_hash=prepared.request_hash,
         datasource_id=datasource.id,
         materialization_id=materialization.id,
-        status="published",
+        status=status,
         details_={
             "fingerprint": prepared.fingerprint,
-            "row_count": len(prepared.rows),
-            "set_current": request.set_current,
+            "row_count": prepared.row_count,
+            "set_current": (
+                request.set_current and status != "review_required"
+            ),
             "frozen": request.freeze,
+            "payload_backend": (
+                prepared.artifact.backend
+                if prepared.artifact is not None
+                else "database_rows"
+            ),
             "validation": materialization.provenance_.get(
                 "publication_validation",
                 {},
@@ -1889,7 +2137,7 @@ def _create_publication_record(
 def _publication_request_hash(
     request: DatasourcePublicationRequest,
     *,
-    normalized: Sequence[Mapping[str, object]],
+    normalized: Sequence[Mapping[str, object]] | None,
     fingerprint: str,
 ) -> str:
     payload = {
@@ -1899,7 +2147,26 @@ def _publication_request_hash(
         "name": request.name,
         "source_name": request.source_name,
         "description": request.description,
-        "rows": [dict(row) for row in normalized],
+        "rows": [dict(row) for row in (normalized or ())],
+        "artifact": (
+            {
+                "backend": request.artifact.backend,
+                "locator": request.artifact.locator,
+                "checksum": request.artifact.checksum.casefold(),
+                "row_count": request.artifact.row_count,
+                "byte_count": request.artifact.byte_count,
+                "schema": [
+                    field_payload(field) for field in request.artifact.schema
+                ],
+                "fingerprint": request.artifact.fingerprint.casefold(),
+                "media_type": request.artifact.media_type,
+                "checkpoint": dict(request.artifact.checkpoint),
+                "metadata": dict(request.artifact.metadata),
+                "validation": dict(request.artifact.validation),
+            }
+            if request.artifact is not None
+            else None
+        ),
         "fingerprint": fingerprint,
         "freeze": request.freeze,
         "frozen_label": request.frozen_label,