Add datasource quality and schema gates

README.md

@@ -23,6 +23,12 @@ Producer modules can append a bounded tabular result or create a new static
 datasource through an idempotent capability. The publication ledger retains the
 producer run, output materialization, provenance, and replay identity.
 
+Tabular staging also evaluates governed quality rules and classifies schema
+changes before promotion. Blocking stages remain inspectable, and successful
+promotion preserves the exact policy hash and validation result in immutable
+materialization provenance. The supported contract is documented in
+[docs/QUALITY_POLICY.md](docs/QUALITY_POLICY.md).
+
 The contracts already model database, HTTP/REST, directory, file, feed,
 document, binary, directory, and stream sources so providers can be added
 without changing consumers. Larger durable artifact-backed publications remain

docs/CONCEPT.md

@@ -76,6 +76,14 @@ A stage records schema, row and byte counts, fingerprint, validation result,
 provenance, and intended target. Promotion either creates a datasource or adds
 a new immutable materialization to an existing compatible datasource.
 
+For bounded tabular stages, the Datasource quality policy now enforces row,
+field, nullability, uniqueness, numeric range, and embedded referential-set
+rules. Updates receive a deterministic compatible, warning, or breaking schema
+classification. Errors and breaking changes keep a stage inspectable but block
+promotion; warnings remain visible and promotable. The policy hash, diagnostics,
+and schema diff are retained with the promoted materialization evidence. See
+[QUALITY_POLICY.md](QUALITY_POLICY.md) for the contract and its privacy bounds.
+
 The first slice stores bounded tabular JSON/CSV stages. Future providers may
 stage file references, object-store blobs, directory snapshots, or streaming
 checkpoints through the same lifecycle contract.

docs/INTERFACE_PATTERN_MIGRATION.md

@@ -9,7 +9,7 @@ editor, previews, and immutable materialization history.
 | Surface | Archetype | Consequence class | Contract |
 | --- | --- | --- | --- |
 | `/datasources` catalogue | Governed directory | Select, register, refresh, freeze, govern, or retire | Shared loading/empty/error, permission, disabled-reason, contextual-help, and read-only states |
-| Staging | Review/preflight queue | Upload, validate, inspect, and promote | Non-consumable bounded stage, explicit promotion confirmation, immutable resulting revision |
+| Staging | Review/preflight queue | Upload, validate, inspect, and promote | Non-consumable bounded stage with privacy-safe quality/schema diagnostics, explicit promotion confirmation, and immutable validation evidence on the resulting revision |
 | Connector origins | Optional-provider directory | Register a live or cached source | Provider availability and supported modes without hard Connectors dependency |
 | Governance editor | Effective authority/provenance editor | Change institutional data context | Guarded draft, authority/source/owner/purpose/quality/freshness semantics |
 | Preview/materializations | Evidence register | Inspect current sample and immutable revisions | Row/schema bounds, freshness, provenance, fingerprints, hashes, and frozen labels |

docs/QUALITY_POLICY.md

@@ -0,0 +1,83 @@
+# Datasource Stage Quality Policy
+
+Datasource quality policy is a deterministic JSON contract stored in
+`governance.quality_policy`. A tabular stage inherits the current target
+Datasource policy unless the stage has its own governed definition. Validation
+runs before the stage is stored, but a failed stage remains available for
+inspection and correction through a new stage.
+
+## Contract
+
+```json
+{
+  "version": "monthly-import-v1",
+  "rules": [
+    {"id": "non-empty", "type": "row_count", "minimum": 1},
+    {"id": "columns", "type": "required_fields", "fields": ["id", "status"]},
+    {"id": "id-shape", "type": "field", "field": "id", "data_type": "integer", "nullable": false},
+    {"id": "id-present", "type": "not_null", "fields": ["id"]},
+    {"id": "id-unique", "type": "unique", "fields": ["id"]},
+    {"id": "amount-range", "type": "range", "field": "amount", "minimum": 0},
+    {"id": "known-status", "type": "referential", "field": "status", "allowed_values": ["new", "closed"]}
+  ],
+  "schema_policy": {
+    "field_added_required": "warning",
+    "field_removed": "breaking"
+  }
+}
+```
+
+Every rule has a stable `id` and may set `severity` to `error` or `warning`.
+Errors block promotion; warnings require review but leave the stage ready. The
+supported rules are:
+
+- `row_count`: optional non-negative `minimum` and `maximum`;
+- `required_fields`: fields that must exist in the detected schema;
+- `field`: expected `data_type` and/or `nullable` contract for one field;
+- `not_null`: one `field` or a `fields` list that must contain no nulls;
+- `unique`: one `field` or a composite `fields` list, with optional
+  `ignore_nulls`;
+- `range`: numeric `minimum` and/or `maximum`, with optional `allow_null`;
+- `referential`: a field and a bounded `allowed_values` set, with optional
+  `allow_null`.
+
+The bounded referential rule is intentionally local and reproducible. A future
+cross-Datasource reference rule must freeze the referenced materialization and
+perform an independent read-authorization check; it must not silently read the
+current state of another protected Datasource.
+
+Use embedded values only for non-sensitive code sets. Quality policy is
+catalogue governance metadata and can be visible to actors who are not allowed
+to read protected rows.
+
+## Schema Classification
+
+When a stage targets an existing Datasource, fields are compared by stable
+name. Defaults are:
+
+| Change | Default |
+| --- | --- |
+| nullable field added | compatible |
+| required field added | warning |
+| field removed | breaking |
+| integer widened to number, or unknown resolved | compatible |
+| other type change | breaking |
+| nullability relaxed | breaking |
+| nullability tightened | compatible |
+| existing field order changed | warning |
+
+Each key may be overridden in `schema_policy` with `compatible`, `warning`, or
+`breaking`. The highest resulting classification is the stage classification.
+A breaking classification blocks promotion.
+
+## Evidence And Privacy
+
+Validation records the policy contract version, a SHA-256 hash of the complete
+policy, rule counts, diagnostics, and the complete schema diff. Row diagnostics
+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.
+
+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

@@ -17,6 +17,7 @@ from govoplan_core.core.module_guards import (
     persistent_table_uninstall_guard,
 )
 from govoplan_core.core.modules import (
+    DocumentationLink,
     DocumentationTopic,
     FrontendModule,
     FrontendRoute,
@@ -381,7 +382,8 @@ manifest = ModuleManifest(
                 "controls the data. The authoritative source, owner, steward, responsible organization/function, schema owner, privacy "
                 "profile, retention policy, transfer agreement, legal basis, holds, correction procedure, purposes, official keys, and "
                 "known limits provide discoverable institutional context. Freshness and quality policies are typed JSON contracts retained "
-                "with materialization evidence; enforcement remains with the provider or consuming control that declares support. Metadata "
+                "with materialization evidence. Datasources enforces the declared bounded tabular stage rules and schema policy; origin-specific "
+                "or consumer-specific controls still remain with the provider or consuming control that declares support. Metadata "
                 "visibility never grants row access."
             ),
             layer="available",
@@ -402,6 +404,44 @@ manifest = ModuleManifest(
                 ],
             },
         ),
+        DocumentationTopic(
+            id="datasources.quality-gates",
+            title="Validate a Datasource stage",
+            summary="Apply deterministic quality rules and schema-change policy before staged 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 "
+                "not inferred from arbitrary JSON flags and remain separate governed lifecycle work."
+            ),
+            layer="available",
+            documentation_types=("admin", "user"),
+            audience=("operator", "module_admin", "data_steward", "power_user"),
+            links=(
+                DocumentationLink(
+                    label="Datasource quality policy",
+                    href="govoplan-datasources/docs/QUALITY_POLICY.md",
+                    kind="repository",
+                ),
+            ),
+            related_modules=("policy", "approvals", "audit", "dataflow", "workflow_engine"),
+            order=72,
+            metadata={
+                "seed": True,
+                "help_contexts": [
+                    "datasources.staging",
+                    "datasources.staging.validation",
+                    "datasources.field.quality-policy",
+                    "datasources.action.promote",
+                ],
+                "limitations": [
+                    "Referential rules currently use a bounded embedded value set rather than reading another protected Datasource.",
+                    "Approval authority and automatic retention execution are not part of the current stage contract.",
+                ],
+            },
+        ),
         DocumentationTopic(
             id="datasources.reference.fields-and-consequences",
             title="Datasource fields and lifecycle consequences",
@@ -419,7 +459,7 @@ manifest = ModuleManifest(
             documentation_types=("admin", "user"),
             audience=("operator", "module_admin", "power_user", "product_owner"),
             related_modules=("connectors", "dataflow", "workflow_engine", "reporting", "audit"),
-            order=72,
+            order=73,
             metadata={
                 "seed": True,
                 "help_contexts": [

src/govoplan_datasources/backend/quality.py

@@ -0,0 +1,654 @@
+from __future__ import annotations
+
+import hashlib
+import json
+from collections.abc import Mapping, Sequence
+from dataclasses import dataclass, field
+from typing import Literal
+
+from govoplan_core.core.datasources import DatasourceField
+
+
+DiagnosticSeverity = Literal["error", "warning"]
+SchemaClassification = Literal["compatible", "warning", "breaking"]
+
+QUALITY_POLICY_VERSION = 1
+MAX_DIAGNOSTIC_ROWS = 25
+MAX_REFERENCE_VALUES = 10_000
+
+
+@dataclass(frozen=True, slots=True)
+class QualityDiagnostic:
+    severity: DiagnosticSeverity
+    code: str
+    message: str
+    rule_id: str | None = None
+    affected_rows: int | None = None
+    row_numbers: tuple[int, ...] = ()
+    details: Mapping[str, object] = field(default_factory=dict)
+
+    def to_dict(self) -> dict[str, object]:
+        payload: dict[str, object] = {
+            "severity": self.severity,
+            "code": self.code,
+            "message": self.message,
+        }
+        if self.rule_id:
+            payload["rule_id"] = self.rule_id
+        if self.affected_rows is not None:
+            payload["affected_rows"] = self.affected_rows
+        if self.row_numbers:
+            payload["row_numbers"] = list(self.row_numbers)
+            payload["row_numbers_truncated"] = self.affected_rows is not None and self.affected_rows > len(self.row_numbers)
+        if self.details:
+            payload["details"] = dict(self.details)
+        return payload
+
+
+@dataclass(frozen=True, slots=True)
+class SchemaChange:
+    code: str
+    classification: SchemaClassification
+    message: str
+    field_name: str | None = None
+    before: Mapping[str, object] | None = None
+    after: Mapping[str, object] | None = None
+
+    def to_dict(self) -> dict[str, object]:
+        payload: dict[str, object] = {
+            "code": self.code,
+            "classification": self.classification,
+            "message": self.message,
+        }
+        if self.field_name:
+            payload["field"] = self.field_name
+        if self.before is not None:
+            payload["before"] = dict(self.before)
+        if self.after is not None:
+            payload["after"] = dict(self.after)
+        return payload
+
+
+_SCHEMA_DEFAULTS: dict[str, SchemaClassification] = {
+    "field_added_nullable": "compatible",
+    "field_added_required": "warning",
+    "field_removed": "breaking",
+    "type_widened": "compatible",
+    "type_changed": "breaking",
+    "nullability_relaxed": "breaking",
+    "nullability_tightened": "compatible",
+    "field_order_changed": "warning",
+}
+_CLASSIFICATION_RANK = {"compatible": 0, "warning": 1, "breaking": 2}
+
+
+def validate_stage(
+    *,
+    rows: Sequence[Mapping[str, object]],
+    schema: Sequence[DatasourceField],
+    quality_policy: Mapping[str, object] | None,
+    baseline_schema: Sequence[DatasourceField] | None = None,
+) -> dict[str, object]:
+    policy = dict(quality_policy or {})
+    diagnostics: list[QualityDiagnostic] = []
+    evaluated_rule_ids: list[str] = []
+    failed_rule_ids: set[str] = set()
+
+    raw_rules = policy.get("rules", [])
+    if not _is_sequence(raw_rules):
+        diagnostics.append(_policy_error("Quality policy rules must be a list."))
+        raw_rules = []
+
+    known_rule_ids: set[str] = set()
+    for index, raw_rule in enumerate(raw_rules):
+        fallback_rule_id = f"rule-{index + 1}"
+        if not isinstance(raw_rule, Mapping):
+            diagnostics.append(_policy_error("Each quality rule must be an object.", fallback_rule_id))
+            failed_rule_ids.add(fallback_rule_id)
+            continue
+        rule_id = str(raw_rule.get("id") or fallback_rule_id).strip()
+        if not rule_id:
+            rule_id = fallback_rule_id
+        if rule_id in known_rule_ids:
+            diagnostics.append(_policy_error(f"Quality rule id {rule_id!r} is duplicated.", rule_id))
+            failed_rule_ids.add(rule_id)
+            continue
+        known_rule_ids.add(rule_id)
+        evaluated_rule_ids.append(rule_id)
+        rule_diagnostics = _evaluate_rule(
+            rule_id=rule_id,
+            rule=raw_rule,
+            rows=rows,
+            schema=schema,
+        )
+        diagnostics.extend(rule_diagnostics)
+        if rule_diagnostics:
+            failed_rule_ids.add(rule_id)
+
+    schema_policy, schema_policy_diagnostics = _schema_policy(policy.get("schema_policy"))
+    diagnostics.extend(schema_policy_diagnostics)
+    schema_changes = (
+        _compare_schemas(baseline_schema, schema, schema_policy)
+        if baseline_schema is not None
+        else ()
+    )
+    schema_classification: str = "new"
+    if baseline_schema is not None:
+        schema_classification = _aggregate_schema_classification(schema_changes)
+        for change in schema_changes:
+            if change.classification == "warning":
+                diagnostics.append(
+                    QualityDiagnostic(
+                        severity="warning",
+                        code=change.code,
+                        message=change.message,
+                        details={"field": change.field_name} if change.field_name else {},
+                    )
+                )
+            elif change.classification == "breaking":
+                diagnostics.append(
+                    QualityDiagnostic(
+                        severity="error",
+                        code=change.code,
+                        message=change.message,
+                        details={"field": change.field_name} if change.field_name else {},
+                    )
+                )
+
+    errors = [item.to_dict() for item in diagnostics if item.severity == "error"]
+    warnings = [item.to_dict() for item in diagnostics if item.severity == "warning"]
+    return {
+        "version": QUALITY_POLICY_VERSION,
+        "policy_version": str(policy.get("version") or QUALITY_POLICY_VERSION),
+        "policy_hash": _policy_hash(policy),
+        "valid": not errors,
+        "errors": errors,
+        "warnings": warnings,
+        "quality": {
+            "rules_evaluated": len(evaluated_rule_ids),
+            "rules_passed": len(evaluated_rule_ids) - len(failed_rule_ids.intersection(evaluated_rule_ids)),
+            "rules_failed": len(failed_rule_ids),
+        },
+        "schema_change": {
+            "classification": schema_classification,
+            "changes": [change.to_dict() for change in schema_changes],
+        },
+    }
+
+
+def _evaluate_rule(
+    *,
+    rule_id: str,
+    rule: Mapping[str, object],
+    rows: Sequence[Mapping[str, object]],
+    schema: Sequence[DatasourceField],
+) -> tuple[QualityDiagnostic, ...]:
+    kind = str(rule.get("type") or "").strip()
+    severity = str(rule.get("severity") or "error").strip()
+    if severity not in {"error", "warning"}:
+        return (_policy_error("Quality rule severity must be error or warning.", rule_id),)
+    if kind == "row_count":
+        return _row_count_rule(rule_id, severity, rule, len(rows))
+    if kind == "required_fields":
+        return _required_fields_rule(rule_id, severity, rule, schema)
+    if kind == "field":
+        return _field_rule(rule_id, severity, rule, schema)
+    if kind == "not_null":
+        return _not_null_rule(rule_id, severity, rule, rows, schema)
+    if kind == "unique":
+        return _unique_rule(rule_id, severity, rule, rows, schema)
+    if kind == "range":
+        return _range_rule(rule_id, severity, rule, rows, schema)
+    if kind == "referential":
+        return _referential_rule(rule_id, severity, rule, rows, schema)
+    return (_policy_error(f"Unsupported quality rule type {kind!r}.", rule_id),)
+
+
+def _row_count_rule(
+    rule_id: str,
+    severity: str,
+    rule: Mapping[str, object],
+    row_count: int,
+) -> tuple[QualityDiagnostic, ...]:
+    minimum = _optional_non_negative_int(rule.get("minimum"))
+    maximum = _optional_non_negative_int(rule.get("maximum"))
+    if rule.get("minimum") is not None and minimum is None:
+        return (_policy_error("A row_count rule minimum must be a non-negative integer.", rule_id),)
+    if rule.get("maximum") is not None and maximum is None:
+        return (_policy_error("A row_count rule maximum must be a non-negative integer.", rule_id),)
+    if minimum is None and maximum is None:
+        return (_policy_error("A row_count rule requires minimum or maximum.", rule_id),)
+    if minimum is not None and maximum is not None and minimum > maximum:
+        return (_policy_error("A row_count rule minimum cannot exceed maximum.", rule_id),)
+    if (minimum is not None and row_count < minimum) or (maximum is not None and row_count > maximum):
+        return (
+            QualityDiagnostic(
+                severity=_severity(severity),
+                code="quality.row_count",
+                message=f"Row count {row_count} is outside the configured bounds.",
+                rule_id=rule_id,
+                details={"actual": row_count, "minimum": minimum, "maximum": maximum},
+            ),
+        )
+    return ()
+
+
+def _required_fields_rule(
+    rule_id: str,
+    severity: str,
+    rule: Mapping[str, object],
+    schema: Sequence[DatasourceField],
+) -> tuple[QualityDiagnostic, ...]:
+    fields, error = _fields_option(rule, "fields")
+    if error:
+        return (_policy_error(error, rule_id),)
+    known = {field.name for field in schema}
+    missing = [name for name in fields if name not in known]
+    if not missing:
+        return ()
+    return (
+        QualityDiagnostic(
+            severity=_severity(severity),
+            code="quality.required_fields",
+            message=f"Required fields are missing: {', '.join(missing)}.",
+            rule_id=rule_id,
+            details={"fields": missing},
+        ),
+    )
+
+
+def _field_rule(
+    rule_id: str,
+    severity: str,
+    rule: Mapping[str, object],
+    schema: Sequence[DatasourceField],
+) -> tuple[QualityDiagnostic, ...]:
+    field_name = _field_option(rule)
+    if not field_name:
+        return (_policy_error("A field rule requires field.", rule_id),)
+    field_value = next((item for item in schema if item.name == field_name), None)
+    if field_value is None:
+        return (
+            QualityDiagnostic(
+                severity=_severity(severity),
+                code="quality.field_missing",
+                message=f"Required field {field_name!r} is missing.",
+                rule_id=rule_id,
+                details={"field": field_name},
+            ),
+        )
+    diagnostics: list[QualityDiagnostic] = []
+    expected_type = str(rule.get("data_type") or "").strip()
+    expected_nullable = rule.get("nullable")
+    if not expected_type and expected_nullable is None:
+        return (_policy_error("A field rule requires data_type or nullable.", rule_id),)
+    if expected_type and field_value.data_type != expected_type:
+        diagnostics.append(
+            QualityDiagnostic(
+                severity=_severity(severity),
+                code="quality.field_type",
+                message=f"Field {field_name!r} has type {field_value.data_type!r}; expected {expected_type!r}.",
+                rule_id=rule_id,
+                details={"field": field_name, "actual": field_value.data_type, "expected": expected_type},
+            )
+        )
+    if expected_nullable is not None and not isinstance(expected_nullable, bool):
+        diagnostics.append(_policy_error("A field rule nullable value must be boolean.", rule_id))
+    elif isinstance(expected_nullable, bool) and field_value.nullable != expected_nullable:
+        diagnostics.append(
+            QualityDiagnostic(
+                severity=_severity(severity),
+                code="quality.field_nullability",
+                message=f"Field {field_name!r} nullability does not match the configured contract.",
+                rule_id=rule_id,
+                details={"field": field_name, "actual": field_value.nullable, "expected": expected_nullable},
+            )
+        )
+    return tuple(diagnostics)
+
+
+def _not_null_rule(
+    rule_id: str,
+    severity: str,
+    rule: Mapping[str, object],
+    rows: Sequence[Mapping[str, object]],
+    schema: Sequence[DatasourceField],
+) -> tuple[QualityDiagnostic, ...]:
+    fields, error = _fields_option(rule, "fields", allow_single_field=True)
+    if error:
+        return (_policy_error(error, rule_id),)
+    missing = _missing_fields(fields, schema)
+    if missing:
+        return (_policy_error(f"not_null references unknown fields: {', '.join(missing)}.", rule_id),)
+    affected = [index + 1 for index, row in enumerate(rows) if any(row.get(name) is None for name in fields)]
+    return _row_diagnostic(
+        severity,
+        "quality.not_null",
+        f"Rows contain null values in required fields: {', '.join(fields)}.",
+        rule_id,
+        affected,
+        {"fields": list(fields)},
+    )
+
+
+def _unique_rule(
+    rule_id: str,
+    severity: str,
+    rule: Mapping[str, object],
+    rows: Sequence[Mapping[str, object]],
+    schema: Sequence[DatasourceField],
+) -> tuple[QualityDiagnostic, ...]:
+    fields, error = _fields_option(rule, "fields", allow_single_field=True)
+    if error:
+        return (_policy_error(error, rule_id),)
+    missing = _missing_fields(fields, schema)
+    if missing:
+        return (_policy_error(f"unique references unknown fields: {', '.join(missing)}.", rule_id),)
+    ignore_nulls = rule.get("ignore_nulls", False)
+    if not isinstance(ignore_nulls, bool):
+        return (_policy_error("A unique rule ignore_nulls value must be boolean.", rule_id),)
+    occurrences: dict[str, list[int]] = {}
+    for index, row in enumerate(rows, start=1):
+        values = tuple(row.get(name) for name in fields)
+        if ignore_nulls and any(value is None for value in values):
+            continue
+        key = json.dumps(values, sort_keys=True, separators=(",", ":"), default=str)
+        occurrences.setdefault(key, []).append(index)
+    affected = sorted(index for indexes in occurrences.values() if len(indexes) > 1 for index in indexes)
+    return _row_diagnostic(
+        severity,
+        "quality.unique",
+        f"Rows violate uniqueness for fields: {', '.join(fields)}.",
+        rule_id,
+        affected,
+        {"fields": list(fields)},
+    )
+
+
+def _range_rule(
+    rule_id: str,
+    severity: str,
+    rule: Mapping[str, object],
+    rows: Sequence[Mapping[str, object]],
+    schema: Sequence[DatasourceField],
+) -> tuple[QualityDiagnostic, ...]:
+    field_name = _field_option(rule)
+    if not field_name:
+        return (_policy_error("A range rule requires field.", rule_id),)
+    if field_name in _missing_fields((field_name,), schema):
+        return (_policy_error(f"range references unknown field {field_name!r}.", rule_id),)
+    minimum = _optional_number(rule.get("minimum"))
+    maximum = _optional_number(rule.get("maximum"))
+    if rule.get("minimum") is not None and minimum is None:
+        return (_policy_error("A range rule minimum must be numeric.", rule_id),)
+    if rule.get("maximum") is not None and maximum is None:
+        return (_policy_error("A range rule maximum must be numeric.", rule_id),)
+    if minimum is None and maximum is None:
+        return (_policy_error("A range rule requires minimum or maximum.", rule_id),)
+    if minimum is not None and maximum is not None and minimum > maximum:
+        return (_policy_error("A range rule minimum cannot exceed maximum.", rule_id),)
+    allow_null = rule.get("allow_null", True)
+    if not isinstance(allow_null, bool):
+        return (_policy_error("A range rule allow_null value must be boolean.", rule_id),)
+    affected: list[int] = []
+    for index, row in enumerate(rows, start=1):
+        value = row.get(field_name)
+        if value is None and allow_null:
+            continue
+        if not _is_number(value) or (minimum is not None and value < minimum) or (maximum is not None and value > maximum):
+            affected.append(index)
+    return _row_diagnostic(
+        severity,
+        "quality.range",
+        f"Rows contain values outside the configured range for {field_name!r}.",
+        rule_id,
+        affected,
+        {"field": field_name, "minimum": minimum, "maximum": maximum},
+    )
+
+
+def _referential_rule(
+    rule_id: str,
+    severity: str,
+    rule: Mapping[str, object],
+    rows: Sequence[Mapping[str, object]],
+    schema: Sequence[DatasourceField],
+) -> tuple[QualityDiagnostic, ...]:
+    field_name = _field_option(rule)
+    if not field_name:
+        return (_policy_error("A referential rule requires field.", rule_id),)
+    if field_name in _missing_fields((field_name,), schema):
+        return (_policy_error(f"referential references unknown field {field_name!r}.", rule_id),)
+    allowed_values = rule.get("allowed_values")
+    if not _is_sequence(allowed_values):
+        return (_policy_error("A bounded referential rule requires allowed_values.", rule_id),)
+    if len(allowed_values) > MAX_REFERENCE_VALUES:
+        return (_policy_error(f"A bounded referential rule supports at most {MAX_REFERENCE_VALUES:,} values.", rule_id),)
+    allow_null = rule.get("allow_null", False)
+    if not isinstance(allow_null, bool):
+        return (_policy_error("A referential rule allow_null value must be boolean.", rule_id),)
+    allowed = {_canonical_value(value) for value in allowed_values}
+    affected = [
+        index
+        for index, row in enumerate(rows, start=1)
+        if not (allow_null and row.get(field_name) is None)
+        and _canonical_value(row.get(field_name)) not in allowed
+    ]
+    return _row_diagnostic(
+        severity,
+        "quality.referential",
+        f"Rows contain unrecognized references in {field_name!r}.",
+        rule_id,
+        affected,
+        {"field": field_name, "allowed_value_count": len(allowed)},
+    )
+
+
+def _compare_schemas(
+    baseline: Sequence[DatasourceField],
+    candidate: Sequence[DatasourceField],
+    policy: Mapping[str, SchemaClassification],
+) -> tuple[SchemaChange, ...]:
+    before_by_name = {field.name: field for field in baseline}
+    after_by_name = {field.name: field for field in candidate}
+    changes: list[SchemaChange] = []
+    for field_name, before in before_by_name.items():
+        after = after_by_name.get(field_name)
+        if after is None:
+            changes.append(_schema_change("field_removed", field_name, before, None, policy))
+            continue
+        if before.data_type != after.data_type:
+            kind = "type_widened" if (before.data_type, after.data_type) in {("integer", "number"), ("unknown", after.data_type)} else "type_changed"
+            changes.append(_schema_change(kind, field_name, before, after, policy))
+        if before.nullable != after.nullable:
+            kind = "nullability_relaxed" if not before.nullable and after.nullable else "nullability_tightened"
+            changes.append(_schema_change(kind, field_name, before, after, policy))
+    for field_name, after in after_by_name.items():
+        if field_name in before_by_name:
+            continue
+        kind = "field_added_nullable" if after.nullable else "field_added_required"
+        changes.append(_schema_change(kind, field_name, None, after, policy))
+    shared_before = [field.name for field in baseline if field.name in after_by_name]
+    shared_after = [field.name for field in candidate if field.name in before_by_name]
+    if shared_before != shared_after:
+        changes.append(
+            SchemaChange(
+                code="schema.field_order_changed",
+                classification=policy["field_order_changed"],
+                message="The order of existing fields changed.",
+            )
+        )
+    return tuple(changes)
+
+
+def _schema_change(
+    kind: str,
+    field_name: str,
+    before: DatasourceField | None,
+    after: DatasourceField | None,
+    policy: Mapping[str, SchemaClassification],
+) -> SchemaChange:
+    before_type = before.data_type if before is not None else "unknown"
+    after_type = after.data_type if after is not None else "unknown"
+    messages = {
+        "field_added_nullable": f"Nullable field {field_name!r} was added.",
+        "field_added_required": f"Required field {field_name!r} was added.",
+        "field_removed": f"Field {field_name!r} was removed.",
+        "type_widened": f"Field {field_name!r} widened from {before_type!r} to {after_type!r}.",
+        "type_changed": f"Field {field_name!r} changed type from {before_type!r} to {after_type!r}.",
+        "nullability_relaxed": f"Field {field_name!r} now permits null values.",
+        "nullability_tightened": f"Field {field_name!r} no longer contains null values.",
+    }
+    return SchemaChange(
+        code=f"schema.{kind}",
+        classification=policy[kind],
+        message=messages[kind],
+        field_name=field_name,
+        before=_field_dict(before),
+        after=_field_dict(after),
+    )
+
+
+def _schema_policy(
+    raw_policy: object,
+) -> tuple[dict[str, SchemaClassification], tuple[QualityDiagnostic, ...]]:
+    result = dict(_SCHEMA_DEFAULTS)
+    if raw_policy is None:
+        return result, ()
+    if not isinstance(raw_policy, Mapping):
+        return result, (_policy_error("schema_policy must be an object."),)
+    diagnostics: list[QualityDiagnostic] = []
+    for key, raw_value in raw_policy.items():
+        if key not in result:
+            diagnostics.append(_policy_error(f"Unsupported schema policy change {key!r}."))
+            continue
+        value = str(raw_value).strip()
+        if value not in _CLASSIFICATION_RANK:
+            diagnostics.append(_policy_error(f"Schema policy {key!r} must be compatible, warning, or breaking."))
+            continue
+        result[str(key)] = value  # type: ignore[assignment]
+    return result, tuple(diagnostics)
+
+
+def _aggregate_schema_classification(changes: Sequence[SchemaChange]) -> SchemaClassification:
+    if not changes:
+        return "compatible"
+    return max((change.classification for change in changes), key=_CLASSIFICATION_RANK.__getitem__)
+
+
+def _row_diagnostic(
+    severity: str,
+    code: str,
+    message: str,
+    rule_id: str,
+    affected: Sequence[int],
+    details: Mapping[str, object],
+) -> tuple[QualityDiagnostic, ...]:
+    if not affected:
+        return ()
+    return (
+        QualityDiagnostic(
+            severity=_severity(severity),
+            code=code,
+            message=message,
+            rule_id=rule_id,
+            affected_rows=len(affected),
+            row_numbers=tuple(affected[:MAX_DIAGNOSTIC_ROWS]),
+            details=details,
+        ),
+    )
+
+
+def _policy_error(message: str, rule_id: str | None = None) -> QualityDiagnostic:
+    return QualityDiagnostic(
+        severity="error",
+        code="quality.policy_invalid",
+        message=message,
+        rule_id=rule_id,
+    )
+
+
+def _fields_option(
+    rule: Mapping[str, object],
+    key: str,
+    *,
+    allow_single_field: bool = False,
+) -> tuple[tuple[str, ...], str | None]:
+    raw_fields = rule.get(key)
+    if raw_fields is None and allow_single_field:
+        raw_fields = [rule.get("field")] if rule.get("field") is not None else None
+    if not _is_sequence(raw_fields):
+        return (), f"Quality rule {key} must be a non-empty list."
+    fields = tuple(str(value).strip() for value in raw_fields)
+    if not fields or any(not value for value in fields):
+        return (), f"Quality rule {key} must be a non-empty list."
+    if len(fields) != len(set(fields)):
+        return (), f"Quality rule {key} cannot contain duplicates."
+    return fields, None
+
+
+def _field_option(rule: Mapping[str, object]) -> str:
+    return str(rule.get("field") or "").strip()
+
+
+def _missing_fields(
+    fields: Sequence[str],
+    schema: Sequence[DatasourceField],
+) -> tuple[str, ...]:
+    known = {field.name for field in schema}
+    return tuple(name for name in fields if name not in known)
+
+
+def _field_dict(value: DatasourceField | None) -> dict[str, object] | None:
+    if value is None:
+        return None
+    return {
+        "name": value.name,
+        "data_type": value.data_type,
+        "nullable": value.nullable,
+    }
+
+
+def _policy_hash(policy: Mapping[str, object]) -> str:
+    encoded = json.dumps(policy, sort_keys=True, separators=(",", ":"), default=str)
+    return hashlib.sha256(encoded.encode("utf-8")).hexdigest()
+
+
+def _canonical_value(value: object) -> str:
+    return json.dumps(value, sort_keys=True, separators=(",", ":"), default=str)
+
+
+def _optional_non_negative_int(value: object) -> int | None:
+    if value is None:
+        return None
+    if isinstance(value, bool) or not isinstance(value, int) or value < 0:
+        return None
+    return value
+
+
+def _optional_number(value: object) -> int | float | None:
+    if value is None:
+        return None
+    return value if _is_number(value) else None
+
+
+def _is_number(value: object) -> bool:
+    return isinstance(value, (int, float)) and not isinstance(value, bool)
+
+
+def _is_sequence(value: object) -> bool:
+    return isinstance(value, Sequence) and not isinstance(value, (str, bytes, bytearray))
+
+
+def _severity(value: str) -> DiagnosticSeverity:
+    return "warning" if value == "warning" else "error"
+
+
+__all__ = [
+    "MAX_DIAGNOSTIC_ROWS",
+    "MAX_REFERENCE_VALUES",
+    "QUALITY_POLICY_VERSION",
+    "QualityDiagnostic",
+    "SchemaChange",
+    "validate_stage",
+]

src/govoplan_datasources/backend/router.py

@@ -1,5 +1,7 @@
 from __future__ import annotations
 
+from collections.abc import Mapping
+
 from fastapi import APIRouter, Depends, HTTPException, Query, status
 from sqlalchemy.orm import Session
 
@@ -224,6 +226,11 @@ def api_create_stage(
             "mode": stage.mode,
             "row_count": stage.row_count,
             "fingerprint": stage.fingerprint,
+            "validation_valid": stage.validation.get("valid"),
+            "quality_policy_hash": stage.validation.get("policy_hash"),
+            "schema_classification": _safe_mapping(
+                stage.validation.get("schema_change")
+            ).get("classification"),
         },
     )
     session.commit()
@@ -262,6 +269,9 @@ def api_promote_stage(
             "materialization_ref": materialization.ref,
             "revision": materialization.revision,
             "frozen": materialization.frozen_at is not None,
+            "quality_policy_hash": _safe_mapping(
+                materialization.provenance.get("stage_validation")
+            ).get("policy_hash"),
         },
     )
     session.commit()
@@ -663,6 +673,10 @@ def _governance(
     return DatasourceGovernance.from_mapping(payload.model_dump())
 
 
+def _safe_mapping(value: object) -> Mapping[str, object]:
+    return value if isinstance(value, Mapping) else {}
+
+
 def _audit(
     session: Session,
     principal: ApiPrincipal,

src/govoplan_datasources/backend/schemas.py

@@ -121,6 +121,48 @@ class DatasourceMaterializationListResponse(BaseModel):
     materializations: list[DatasourceMaterializationResponse]
 
 
+class DatasourceValidationDiagnosticResponse(BaseModel):
+    severity: Literal["error", "warning"]
+    code: str
+    message: str
+    rule_id: str | None = None
+    affected_rows: int | None = None
+    row_numbers: list[int] = Field(default_factory=list)
+    row_numbers_truncated: bool = False
+    details: dict[str, Any] = Field(default_factory=dict)
+
+
+class DatasourceSchemaChangeResponse(BaseModel):
+    code: str
+    classification: Literal["compatible", "warning", "breaking"]
+    message: str
+    field: str | None = None
+    before: DatasourceFieldResponse | None = None
+    after: DatasourceFieldResponse | None = None
+
+
+class DatasourceQualitySummaryResponse(BaseModel):
+    rules_evaluated: int = 0
+    rules_passed: int = 0
+    rules_failed: int = 0
+
+
+class DatasourceSchemaChangeSummaryResponse(BaseModel):
+    classification: Literal["new", "compatible", "warning", "breaking"] = "new"
+    changes: list[DatasourceSchemaChangeResponse] = Field(default_factory=list)
+
+
+class DatasourceStageValidationResponse(BaseModel):
+    version: int = 1
+    policy_version: str = "1"
+    policy_hash: str = ""
+    valid: bool = True
+    errors: list[DatasourceValidationDiagnosticResponse] = Field(default_factory=list)
+    warnings: list[DatasourceValidationDiagnosticResponse] = Field(default_factory=list)
+    quality: DatasourceQualitySummaryResponse = Field(default_factory=DatasourceQualitySummaryResponse)
+    schema_change: DatasourceSchemaChangeSummaryResponse = Field(default_factory=DatasourceSchemaChangeSummaryResponse)
+
+
 class DatasourceStageResponse(BaseModel):
     ref: str
     name: str
@@ -134,7 +176,7 @@ class DatasourceStageResponse(BaseModel):
     fields: list[DatasourceFieldResponse] = Field(alias="schema")
     row_count: int | None
     byte_count: int | None
-    validation: dict[str, Any]
+    validation: DatasourceStageValidationResponse
     created_at: str | None
     promoted_at: str | None
     promoted_materialization_ref: str | None

src/govoplan_datasources/backend/service.py

@@ -48,6 +48,7 @@ from govoplan_datasources.backend.payloads import (
     payload_for_materialization,
     validate_payload_size,
 )
+from govoplan_datasources.backend.quality import validate_stage
 from govoplan_datasources.backend.tabular import (
     MAX_READ_ROWS,
     MAX_STAGE_ROWS,
@@ -427,6 +428,12 @@ class SqlDatasourceProvider:
         rows = normalize_rows(stage.rows)
         schema = infer_schema(rows)
         fingerprint = fingerprint_rows(rows, schema)
+        validation = validate_stage(
+            rows=rows,
+            schema=schema,
+            quality_policy=governance.quality_policy,
+            baseline_schema=_fields(target.schema_) if target is not None else None,
+        )
         item = DatasourceStageRecord(
             tenant_id=api_principal.tenant_id,
             target_datasource_id=target.id if target else None,
@@ -436,7 +443,7 @@ class SqlDatasourceProvider:
             kind=stage.kind,
             mode=stage.mode,
             shape=stage.shape,
-            state="ready",
+            state="ready" if validation["valid"] else "invalid",
             provider=_clean_optional(stage.provider),
             provider_ref=_clean_optional(stage.provider_ref),
             schema_=[field_payload(field) for field in schema],
@@ -444,7 +451,7 @@ class SqlDatasourceProvider:
             fingerprint=fingerprint,
             row_count=len(rows),
             byte_count=encoded_size(rows),
-            validation_={"valid": True, "errors": [], "warnings": []},
+            validation_=validation,
             provenance_=dict(stage.provenance),
             metadata_=dict(stage.metadata),
             governance_=governance.to_dict(),
@@ -469,6 +476,10 @@ class SqlDatasourceProvider:
             tenant_id=api_principal.tenant_id,
             stage_ref=stage_ref,
         )
+        if stage.validation_.get("valid") is not True:
+            raise DatasourceValidationError(
+                "The stage has blocking quality or schema diagnostics and cannot be promoted."
+            )
         if stage.state != "ready":
             raise DatasourceValidationError("Only ready stages can be promoted.")
         datasource = (
@@ -538,6 +549,7 @@ class SqlDatasourceProvider:
             provenance={
                 **dict(stage.provenance_),
                 "stage_ref": _stage_ref(stage.id),
+                "stage_validation": dict(stage.validation_),
             },
             metadata=dict(stage.metadata_),
             set_current=True,

tests/test_interface_documentation_contract.py

@@ -26,10 +26,14 @@ class DatasourcesInterfaceDocumentationContractTests(unittest.TestCase):
         topics = {topic.id: topic for topic in manifest.documentation}
         lifecycle = topics["datasources.lifecycle"]
         governance = topics["datasources.governance"]
+        quality = topics["datasources.quality-gates"]
         reference = topics["datasources.reference.fields-and-consequences"]
 
         self.assertIn("datasources.staging", lifecycle.metadata["help_contexts"])
         self.assertIn("datasources.field.authority-mode", governance.metadata["help_contexts"])
+        self.assertIn("datasources.staging.validation", quality.metadata["help_contexts"])
+        self.assertIn("policy version and hash", quality.body)
+        self.assertTrue(quality.metadata["limitations"])
         self.assertIn("datasources.action.promote", reference.metadata["help_contexts"])
         self.assertIn("freeze", reference.metadata["consequence_classes"])
         self.assertIn("retire", reference.metadata["consequence_classes"])

tests/test_lifecycle.py

@@ -265,6 +265,96 @@ class DatasourceLifecycleTests(unittest.TestCase):
         self.assertEqual(first_record.payload_id, frozen_record.payload_id)
         self.assertEqual([], first_record.rows)
 
+    def test_stage_quality_and_schema_gates_block_only_error_diagnostics(self) -> None:
+        governance = DatasourceGovernance(
+            authority_mode="native_authoritative",
+            publication_state="internal",
+            quality_policy={
+                "version": "monthly-cases-v1",
+                "rules": [
+                    {"id": "unique-id", "type": "unique", "fields": ["id"]},
+                ],
+                "schema_policy": {"field_added_required": "warning"},
+            },
+        )
+        first_stage = self.provider.create_stage(
+            self.session,
+            principal(),
+            stage=DatasourceStageInput(
+                name="Monthly cases",
+                source_name="monthly_quality_cases",
+                kind="upload",
+                mode="static",
+                shape="tabular",
+                rows=({"id": 1},),
+                governance=governance,
+            ),
+        )
+        datasource, _first = self.provider.promote_stage(
+            self.session,
+            principal(),
+            stage_ref=first_stage.ref,
+        )
+
+        blocked_stage = self.provider.create_stage(
+            self.session,
+            principal(),
+            stage=DatasourceStageInput(
+                name="Monthly cases",
+                source_name="monthly_quality_cases",
+                kind="upload",
+                mode="static",
+                shape="tabular",
+                target_datasource_ref=datasource.ref,
+                rows=({"id": 2, "name": "Ada"}, {"id": 2, "name": "Lin"}),
+            ),
+        )
+        self.assertEqual("invalid", blocked_stage.state)
+        self.assertFalse(blocked_stage.validation["valid"])
+        self.assertEqual(
+            "quality.unique",
+            blocked_stage.validation["errors"][0]["code"],
+        )
+        self.assertEqual(
+            "warning",
+            blocked_stage.validation["schema_change"]["classification"],
+        )
+        with self.assertRaisesRegex(
+            DatasourceValidationError,
+            "blocking quality or schema diagnostics",
+        ):
+            self.provider.promote_stage(
+                self.session,
+                principal(),
+                stage_ref=blocked_stage.ref,
+            )
+
+        ready_stage = self.provider.create_stage(
+            self.session,
+            principal(),
+            stage=DatasourceStageInput(
+                name="Monthly cases",
+                source_name="monthly_quality_cases",
+                kind="upload",
+                mode="static",
+                shape="tabular",
+                target_datasource_ref=datasource.ref,
+                rows=({"id": 2, "name": "Ada"}, {"id": 3, "name": "Lin"}),
+            ),
+        )
+        self.assertEqual("ready", ready_stage.state)
+        self.assertTrue(ready_stage.validation["valid"])
+        self.assertEqual("schema.field_added_required", ready_stage.validation["warnings"][0]["code"])
+        _updated, materialization = self.provider.promote_stage(
+            self.session,
+            principal(),
+            stage_ref=ready_stage.ref,
+        )
+        self.assertEqual(
+            ready_stage.validation["policy_hash"],
+            materialization.provenance["stage_validation"]["policy_hash"],
+        )
+
     def test_governance_is_queryable_and_snapshotted_per_materialization(self) -> None:
         stage = self.provider.create_stage(
             self.session,

tests/test_quality.py

@@ -0,0 +1,142 @@
+from __future__ import annotations
+
+import unittest
+
+from govoplan_core.core.datasources import DatasourceField
+from govoplan_datasources.backend.quality import MAX_DIAGNOSTIC_ROWS, validate_stage
+from govoplan_datasources.backend.schemas import DatasourceStageValidationResponse
+
+
+class DatasourceQualityTests(unittest.TestCase):
+    def test_quality_rules_report_counts_without_exposing_values(self) -> None:
+        rows = (
+            {"id": 1, "name": "Ada", "amount": 10, "status": "new"},
+            {"id": 1, "name": None, "amount": -2, "status": "unknown"},
+        )
+        schema = (
+            DatasourceField("id", "integer", False),
+            DatasourceField("name", "string", True),
+            DatasourceField("amount", "integer", False),
+            DatasourceField("status", "string", False),
+        )
+        result = validate_stage(
+            rows=rows,
+            schema=schema,
+            quality_policy={
+                "version": "case-import-v2",
+                "rules": [
+                    {"id": "rows", "type": "row_count", "minimum": 3},
+                    {"id": "columns", "type": "required_fields", "fields": ["id", "missing"]},
+                    {"id": "id-shape", "type": "field", "field": "id", "data_type": "string"},
+                    {"id": "names", "type": "not_null", "fields": ["name"]},
+                    {"id": "ids", "type": "unique", "fields": ["id"]},
+                    {"id": "amount", "type": "range", "field": "amount", "minimum": 0},
+                    {
+                        "id": "status",
+                        "type": "referential",
+                        "field": "status",
+                        "allowed_values": ["new", "closed"],
+                    },
+                ],
+            },
+        )
+
+        self.assertFalse(result["valid"])
+        self.assertEqual("case-import-v2", result["policy_version"])
+        errors = result["errors"]
+        self.assertEqual(
+            {
+                "quality.row_count",
+                "quality.required_fields",
+                "quality.field_type",
+                "quality.not_null",
+                "quality.unique",
+                "quality.range",
+                "quality.referential",
+            },
+            {item["code"] for item in errors},
+        )
+        unique = next(item for item in errors if item["code"] == "quality.unique")
+        self.assertEqual(2, unique["affected_rows"])
+        self.assertEqual([1, 2], unique["row_numbers"])
+        self.assertNotIn("unknown", str(errors))
+        contract = DatasourceStageValidationResponse.model_validate(result)
+        self.assertEqual("case-import-v2", contract.policy_version)
+        self.assertEqual(7, contract.quality.rules_failed)
+
+    def test_warning_rules_do_not_block_promotion_readiness(self) -> None:
+        result = validate_stage(
+            rows=({"id": 1},),
+            schema=(DatasourceField("id", "integer", False),),
+            quality_policy={
+                "rules": [
+                    {
+                        "id": "large-batch",
+                        "type": "row_count",
+                        "minimum": 2,
+                        "severity": "warning",
+                    }
+                ]
+            },
+        )
+
+        self.assertTrue(result["valid"])
+        self.assertEqual([], result["errors"])
+        self.assertEqual("quality.row_count", result["warnings"][0]["code"])
+
+    def test_schema_changes_are_classified_and_can_be_governed(self) -> None:
+        baseline = (
+            DatasourceField("id", "integer", False),
+            DatasourceField("label", "string", False),
+        )
+        candidate = (
+            DatasourceField("id", "number", False),
+            DatasourceField("note", "string", True),
+        )
+
+        blocked = validate_stage(
+            rows=(),
+            schema=candidate,
+            baseline_schema=baseline,
+            quality_policy={},
+        )
+        self.assertFalse(blocked["valid"])
+        self.assertEqual("breaking", blocked["schema_change"]["classification"])
+        self.assertEqual(
+            {"schema.type_widened", "schema.field_removed", "schema.field_added_nullable"},
+            {item["code"] for item in blocked["schema_change"]["changes"]},
+        )
+
+        governed = validate_stage(
+            rows=(),
+            schema=candidate,
+            baseline_schema=baseline,
+            quality_policy={"schema_policy": {"field_removed": "warning"}},
+        )
+        self.assertTrue(governed["valid"])
+        self.assertEqual("warning", governed["schema_change"]["classification"])
+
+    def test_malformed_policy_fails_closed_and_hash_is_stable(self) -> None:
+        policy = {"rules": [{"id": "bad", "type": "unique", "fields": []}]}
+        first = validate_stage(rows=(), schema=(), quality_policy=policy)
+        second = validate_stage(rows=(), schema=(), quality_policy=policy)
+
+        self.assertFalse(first["valid"])
+        self.assertEqual("quality.policy_invalid", first["errors"][0]["code"])
+        self.assertEqual(first["policy_hash"], second["policy_hash"])
+
+    def test_diagnostic_row_numbers_are_bounded(self) -> None:
+        rows = tuple({"id": None} for _ in range(MAX_DIAGNOSTIC_ROWS + 10))
+        result = validate_stage(
+            rows=rows,
+            schema=(DatasourceField("id", "unknown", True),),
+            quality_policy={"rules": [{"id": "id", "type": "not_null", "field": "id"}]},
+        )
+        error = result["errors"][0]
+        self.assertEqual(MAX_DIAGNOSTIC_ROWS + 10, error["affected_rows"])
+        self.assertEqual(MAX_DIAGNOSTIC_ROWS, len(error["row_numbers"]))
+        self.assertTrue(error["row_numbers_truncated"])
+
+
+if __name__ == "__main__":
+    unittest.main()

webui/src/api/datasources.ts

@@ -87,6 +87,44 @@ export type DatasourceMaterialization = {
   governance: DatasourceGovernance;
 };
 
+export type DatasourceValidationDiagnostic = {
+  severity: "error" | "warning";
+  code: string;
+  message: string;
+  rule_id?: string;
+  affected_rows?: number;
+  row_numbers?: number[];
+  row_numbers_truncated?: boolean;
+  details?: Record<string, unknown>;
+};
+
+export type DatasourceSchemaChange = {
+  code: string;
+  classification: "compatible" | "warning" | "breaking";
+  message: string;
+  field?: string;
+  before?: DatasourceField;
+  after?: DatasourceField;
+};
+
+export type DatasourceStageValidation = {
+  version?: number;
+  policy_version?: string;
+  policy_hash?: string;
+  valid?: boolean;
+  errors?: DatasourceValidationDiagnostic[];
+  warnings?: DatasourceValidationDiagnostic[];
+  quality?: {
+    rules_evaluated?: number;
+    rules_passed?: number;
+    rules_failed?: number;
+  };
+  schema_change?: {
+    classification?: "new" | "compatible" | "warning" | "breaking";
+    changes?: DatasourceSchemaChange[];
+  };
+};
+
 export type DatasourceStage = {
   ref: string;
   name: string;
@@ -100,7 +138,7 @@ export type DatasourceStage = {
   schema: DatasourceField[];
   row_count?: number | null;
   byte_count?: number | null;
-  validation: Record<string, unknown>;
+  validation: DatasourceStageValidation;
   created_at?: string | null;
   promoted_at?: string | null;
   promoted_materialization_ref?: string | null;

webui/src/features/datasources/DatasourcesPage.tsx

@@ -56,7 +56,9 @@ import {
   type DatasourceMaterialization,
   type DatasourceOrigin,
   type DatasourcePreview,
-  type DatasourceStage
+  type DatasourceSchemaChange,
+  type DatasourceStage,
+  type DatasourceValidationDiagnostic
 } from "../../api/datasources";
 import {
   DATASOURCE_FIELDS_DOCUMENTATION,
@@ -717,6 +719,10 @@ function DatasourceDetail({
 }
 
 function StageDetail({ stage }: { stage: DatasourceStage }) {
+  const errors = stage.validation.errors ?? [];
+  const warnings = stage.validation.warnings ?? [];
+  const schemaChanges = stage.validation.schema_change?.changes ?? [];
+  const schemaClassification = stage.validation.schema_change?.classification ?? "new";
   return (
     <>
       <div className="datasources-metrics">
@@ -738,8 +744,44 @@ function StageDetail({ stage }: { stage: DatasourceStage }) {
           <span><small>Fingerprint</small><strong>{shortFingerprint(stage.fingerprint)}</strong></span>
           <span><small>Target</small><strong>{stage.target_datasource_ref || "New datasource"}</strong></span>
           <span><small>Promoted revision</small><strong>{stage.promoted_materialization_ref || "Not promoted"}</strong></span>
+          <span><small>Quality policy</small><strong>{stage.validation.policy_version || "Local default"}</strong></span>
+          <span><small>Policy hash</small><strong>{shortFingerprint(stage.validation.policy_hash || "")}</strong></span>
+          <span><small>Schema change</small><strong>{readableToken(schemaClassification)}</strong></span>
         </div>
+        {errors.length || warnings.length ? (
+          <div className="datasources-validation-alerts">
+            {errors.length ? (
+              <DismissibleAlert tone="danger" compact dismissible={false}>
+                <strong>Promotion blockers</strong>
+                <ValidationDiagnosticList diagnostics={errors} />
+              </DismissibleAlert>
+            ) : null}
+            {warnings.length ? (
+              <DismissibleAlert tone="warning" compact dismissible={false}>
+                <strong>Review warnings</strong>
+                <ValidationDiagnosticList diagnostics={warnings} />
+              </DismissibleAlert>
+            ) : null}
+          </div>
+        ) : (
+          <div className="datasources-validation-ok">
+            All configured quality rules passed and no blocking schema change was detected.
+          </div>
+        )}
       </section>
+      {schemaChanges.length ? (
+        <section className="datasources-detail-section">
+          <div className="datasources-section-heading">
+            <span>Schema comparison</span>
+            <StatusBadge status={schemaClassification} label={readableToken(schemaClassification)} />
+          </div>
+          <ul className="datasources-schema-changes">
+            {schemaChanges.map((change, index) => (
+              <SchemaChangeItem key={`${change.code}-${change.field ?? index}`} change={change} />
+            ))}
+          </ul>
+        </section>
+      ) : null}
       <section className="datasources-detail-section">
         <div className="datasources-section-heading">
           <span>Detected schema</span>
@@ -751,6 +793,42 @@ function StageDetail({ stage }: { stage: DatasourceStage }) {
   );
 }
 
+function ValidationDiagnosticList({
+  diagnostics
+}: {
+  diagnostics: DatasourceValidationDiagnostic[];
+}) {
+  return (
+    <ul className="datasources-diagnostic-list">
+      {diagnostics.map((diagnostic, index) => {
+        const details = [
+          diagnostic.rule_id ? `Rule ${diagnostic.rule_id}` : "",
+          diagnostic.affected_rows !== undefined ? `${diagnostic.affected_rows} affected row${diagnostic.affected_rows === 1 ? "" : "s"}` : "",
+          diagnostic.row_numbers?.length ? `Rows ${diagnostic.row_numbers.join(", ")}${diagnostic.row_numbers_truncated ? ", …" : ""}` : ""
+        ].filter(Boolean);
+        return (
+          <li key={`${diagnostic.code}-${diagnostic.rule_id ?? index}`}>
+            <span>{diagnostic.message}</span>
+            {details.length ? <small>{details.join(" · ")}</small> : null}
+          </li>
+        );
+      })}
+    </ul>
+  );
+}
+
+function SchemaChangeItem({ change }: { change: DatasourceSchemaChange }) {
+  return (
+    <li>
+      <StatusBadge status={change.classification} label={readableToken(change.classification)} />
+      <span>
+        <strong>{change.message}</strong>
+        {change.field ? <small>{change.field}</small> : null}
+      </span>
+    </li>
+  );
+}
+
 function OriginDetail({ origin }: { origin: DatasourceOrigin }) {
   return (
     <>

webui/src/i18n/generatedTranslations.ts

@@ -62,7 +62,15 @@ const en = {
   "Publication state": "Publication state",
   "Semantic definition": "Semantic definition",
   "Freshness policy (JSON)": "Freshness policy (JSON)",
-  "Quality policy (JSON)": "Quality policy (JSON)"
+  "Quality policy (JSON)": "Quality policy (JSON)",
+  "Quality policy": "Quality policy",
+  "Policy hash": "Policy hash",
+  "Schema change": "Schema change",
+  "Local default": "Local default",
+  "Promotion blockers": "Promotion blockers",
+  "Review warnings": "Review warnings",
+  "All configured quality rules passed and no blocking schema change was detected.": "All configured quality rules passed and no blocking schema change was detected.",
+  "Schema comparison": "Schema comparison"
 } as const;
 
 const de: Record<keyof typeof en, string> = {
@@ -127,7 +135,15 @@ const de: Record<keyof typeof en, string> = {
   "Publication state": "Veröffentlichungsstatus",
   "Semantic definition": "Semantische Definition",
   "Freshness policy (JSON)": "Aktualitätsrichtlinie (JSON)",
-  "Quality policy (JSON)": "Qualitätsrichtlinie (JSON)"
+  "Quality policy (JSON)": "Qualitätsrichtlinie (JSON)",
+  "Quality policy": "Qualitätsrichtlinie",
+  "Policy hash": "Richtlinien-Hash",
+  "Schema change": "Schemaänderung",
+  "Local default": "Lokaler Standard",
+  "Promotion blockers": "Übernahmehindernisse",
+  "Review warnings": "Prüfhinweise",
+  "All configured quality rules passed and no blocking schema change was detected.": "Alle konfigurierten Qualitätsregeln wurden erfüllt und es wurde keine blockierende Schemaänderung erkannt.",
+  "Schema comparison": "Schemavergleich"
 };
 
 export const generatedTranslations: PlatformTranslations = { en, de };

webui/src/styles/datasources.css

@@ -369,6 +369,74 @@
   font-weight: 600;
 }
 
+.datasources-validation-alerts {
+  display: grid;
+  gap: 8px;
+  padding: 10px;
+  border-top: var(--border-line);
+}
+
+.datasources-validation-alerts .alert {
+  margin: 0;
+}
+
+.datasources-diagnostic-list,
+.datasources-schema-changes {
+  display: grid;
+  gap: 7px;
+  margin: 7px 0 0;
+  padding: 0;
+  list-style: none;
+}
+
+.datasources-diagnostic-list li,
+.datasources-diagnostic-list span,
+.datasources-diagnostic-list small,
+.datasources-schema-changes li,
+.datasources-schema-changes span,
+.datasources-schema-changes strong,
+.datasources-schema-changes small {
+  display: block;
+  min-width: 0;
+}
+
+.datasources-diagnostic-list small,
+.datasources-schema-changes small {
+  margin-top: 2px;
+  color: var(--muted);
+  font-size: 11px;
+}
+
+.datasources-validation-ok {
+  padding: 10px 12px;
+  border-top: var(--border-line);
+  color: var(--success);
+  font-size: 12px;
+}
+
+.datasources-schema-changes {
+  margin: 0;
+  padding: 8px 10px;
+}
+
+.datasources-schema-changes li {
+  display: grid;
+  grid-template-columns: auto minmax(0, 1fr);
+  align-items: start;
+  gap: 9px;
+  padding: 6px 0;
+}
+
+.datasources-schema-changes li + li {
+  border-top: var(--border-line);
+}
+
+.datasources-schema-changes strong {
+  color: var(--text-strong);
+  font-size: 12px;
+  font-weight: 600;
+}
+
 .datasources-workspace-empty {
   display: grid;
   place-items: center;