feat(datasources): enforce policy-aware data visibility

src/govoplan_datasources/__init__.py

@@ -1,3 +1,3 @@
 """GovOPlaN Datasources module."""
 
-__version__ = "0.1.19"
+__version__ = "0.1.20"

src/govoplan_datasources/backend/db/models.py

@@ -109,6 +109,10 @@ class DatasourceRecord(Base, TimestampMixin):
     )
     privacy_profile_ref: Mapped[str | None] = mapped_column(String(500), nullable=True)
     retention_policy_ref: Mapped[str | None] = mapped_column(String(500), nullable=True)
+    access_policy_ref: Mapped[str | None] = mapped_column(String(500), nullable=True)
+    visibility_policy: Mapped[dict[str, Any]] = mapped_column(
+        JSON, default=dict, nullable=False
+    )
     hold_refs: Mapped[list[str]] = mapped_column(JSON, default=list, nullable=False)
     publication_state: Mapped[str] = mapped_column(
         String(50), default="draft", nullable=False, index=True

src/govoplan_datasources/backend/manifest.py

@@ -12,6 +12,7 @@ from govoplan_core.core.datasources import (
     CAPABILITY_DATASOURCE_LIFECYCLE,
     CAPABILITY_DATASOURCE_ORIGINS,
     CAPABILITY_DATASOURCE_PUBLICATION,
+    CAPABILITY_POLICY_DATASOURCE_VISIBILITY,
     datasource_artifact_backend_provider,
 )
 from govoplan_core.core.module_guards import (
@@ -62,7 +63,7 @@ from govoplan_datasources.backend.payloads import ExternalArtifactPayloadBackend
 
 MODULE_ID = "datasources"
 MODULE_NAME = "Datasources"
-MODULE_VERSION = "0.1.19"
+MODULE_VERSION = "0.1.20"
 DATASOURCE_INTERFACE_VERSION = "0.2.0"
 
 ARCHITECTURE = ModuleArchitectureDeclaration(
@@ -256,6 +257,7 @@ manifest = ModuleManifest(
         CAPABILITY_AUTH_PERMISSION_EVALUATOR,
         CAPABILITY_DATASOURCE_ORIGINS,
         CAPABILITY_DATASOURCE_ARTIFACT_BACKENDS,
+        CAPABILITY_POLICY_DATASOURCE_VISIBILITY,
     ),
     provides_interfaces=(
         ModuleInterfaceProvider(
@@ -293,6 +295,12 @@ manifest = ModuleManifest(
             version_max_exclusive="2.0.0",
             optional=True,
         ),
+        ModuleInterfaceRequirement(
+            name="policy.datasource_visibility",
+            version_min="1.0.0",
+            version_max_exclusive="2.0.0",
+            optional=True,
+        ),
     ),
     permissions=PERMISSIONS,
     role_templates=ROLE_TEMPLATES,
@@ -547,10 +555,40 @@ manifest = ModuleManifest(
                     "datasources.field.freshness-policy",
                     "datasources.field.quality-policy",
                     "datasources.field.retention-policy",
+                    "datasources.field.access-policy",
+                    "datasources.field.visibility-policy",
                     "datasources.field.transfer-agreement",
                 ],
             },
         ),
+        DocumentationTopic(
+            id="datasources.visibility",
+            title="Datasource and field visibility",
+            summary="Apply source, materialization, field, and row controls before protected rows leave Datasources.",
+            body=(
+                "A local visibility policy can restrict discovery and reads by account, membership, identity, group, role, service account, or authentication method. "
+                "Materialization ACLs additionally protect current and frozen revisions. Field rules classify a field and either replace its value with null or omit the field unless an allow ACL matches. Row filters compare a configured field with a principal account, membership, identity, service-account, group, role, or function-assignment claim; all configured filters and all policy overlays must allow a row. "
+                "Datasources applies these controls before rows leave its provider boundary and returns an opaque permitted-view fingerprint bound to the base revision, effective policies, and relevant principal facts. Denied and filtered reads emit audit evidence containing outcomes and policy diagnostics, never protected values, provider locators, arbitrary metadata, or provenance bodies. "
+                "An optional access-policy reference delegates hierarchical overlays to Policy. Policy can only tighten local behavior. An unresolved reference, unavailable configured provider, or malformed policy fails closed. Without Policy and without a reference, the local visibility policy remains fully usable. Frozen reads enforce the current local policy together with a distinct restrictive policy captured in the frozen governance snapshot."
+            ),
+            layer="available",
+            documentation_types=("admin", "user"),
+            audience=("operator", "module_admin", "data_steward", "auditor"),
+            related_modules=("policy", "access", "audit", "connectors"),
+            order=72,
+            metadata={
+                "seed": True,
+                "help_contexts": [
+                    "datasources.field.access-policy",
+                    "datasources.field.visibility-policy",
+                    "datasources.preview",
+                ],
+                "limitations": [
+                    "Row filters support deterministic equality or membership against principal claims; arbitrary expressions are not executed.",
+                    "Live filtered previews scan at most 10,000 origin rows and disclose a structured limitation diagnostic if the origin is larger.",
+                ],
+            },
+        ),
         DocumentationTopic(
             id="datasources.quality-gates",
             title="Validate staged and produced Datasource revisions",
@@ -585,7 +623,7 @@ manifest = ModuleManifest(
                 "dataflow",
                 "workflow_engine",
             ),
-            order=72,
+            order=73,
             metadata={
                 "seed": True,
                 "help_contexts": [
@@ -624,7 +662,7 @@ manifest = ModuleManifest(
                 "reporting",
                 "audit",
             ),
-            order=73,
+            order=74,
             metadata={
                 "seed": True,
                 "help_contexts": [

src/govoplan_datasources/backend/migrations/versions/c9e3a6f1d4b8_v0120_datasource_visibility.py

@@ -0,0 +1,38 @@
+"""v0.1.20 datasource visibility policy
+
+Revision ID: c9e3a6f1d4b8
+Revises: b8d2f5a0c3e7
+Create Date: 2026-08-21 20:15:00.000000
+"""
+
+from __future__ import annotations
+
+from alembic import op
+import sqlalchemy as sa
+
+
+revision = "c9e3a6f1d4b8"
+down_revision = "b8d2f5a0c3e7"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.add_column(
+        "datasource_catalogue",
+        sa.Column("access_policy_ref", sa.String(length=500), nullable=True),
+    )
+    op.add_column(
+        "datasource_catalogue",
+        sa.Column(
+            "visibility_policy",
+            sa.JSON(),
+            server_default=sa.text("'{}'"),
+            nullable=False,
+        ),
+    )
+
+
+def downgrade() -> None:
+    op.drop_column("datasource_catalogue", "visibility_policy")
+    op.drop_column("datasource_catalogue", "access_policy_ref")

src/govoplan_datasources/backend/router.py

@@ -1,5 +1,7 @@
 from __future__ import annotations
 
+import hashlib
+import json
 from collections.abc import Mapping
 
 from fastapi import APIRouter, Depends, HTTPException, Query, status
@@ -370,7 +372,17 @@ def api_update_datasource_governance(
         action="datasources.governance.updated",
         object_type="datasource",
         object_id=item.ref,
-        details={"governance": item.governance.to_dict()},
+        details={
+            "classification": item.governance.classification,
+            "publication_state": item.governance.publication_state,
+            "access_policy_ref": item.governance.access_policy_ref,
+            "visibility_policy_configured": bool(
+                item.governance.visibility_policy
+            ),
+            "visibility_policy_hash": _mapping_hash(
+                item.governance.visibility_policy
+            ),
+        },
     )
     session.commit()
     return _datasource_response(item)
@@ -580,6 +592,7 @@ def _datasource_response(item: DatasourceDescriptor) -> DatasourceResponse:
                 name=field.name,
                 data_type=field.data_type,
                 nullable=field.nullable,
+                classification=field.classification,
             )
             for field in item.schema
         ],
@@ -610,6 +623,7 @@ def _materialization_response(
                 name=field.name,
                 data_type=field.data_type,
                 nullable=field.nullable,
+                classification=field.classification,
             )
             for field in item.schema
         ],
@@ -643,6 +657,7 @@ def _stage_response(item: DatasourceStage) -> DatasourceStageResponse:
                 name=field.name,
                 data_type=field.data_type,
                 nullable=field.nullable,
+                classification=field.classification,
             )
             for field in item.schema
         ],
@@ -673,6 +688,7 @@ def _origin_response(item: DatasourceOrigin) -> DatasourceOriginResponse:
                 name=field.name,
                 data_type=field.data_type,
                 nullable=field.nullable,
+                classification=field.classification,
             )
             for field in item.schema
         ],
@@ -715,6 +731,13 @@ def _safe_mapping(value: object) -> Mapping[str, object]:
     return value if isinstance(value, Mapping) else {}
 
 
+def _mapping_hash(value: Mapping[str, object]) -> str | None:
+    if not value:
+        return None
+    encoded = json.dumps(value, sort_keys=True, separators=(",", ":"), default=str)
+    return hashlib.sha256(encoded.encode("utf-8")).hexdigest()
+
+
 def _audit(
     session: Session,
     principal: ApiPrincipal,

src/govoplan_datasources/backend/schemas.py

@@ -43,6 +43,8 @@ class DatasourceGovernancePayload(BaseModel):
     classification: str = Field(default="internal", min_length=1, max_length=80)
     privacy_profile_ref: str | None = Field(default=None, max_length=500)
     retention_policy_ref: str | None = Field(default=None, max_length=500)
+    access_policy_ref: str | None = Field(default=None, max_length=500)
+    visibility_policy: dict[str, Any] = Field(default_factory=dict)
     hold_refs: list[str] = Field(default_factory=list, max_length=100)
     publication_state: str = Field(default="draft", min_length=1, max_length=50)
     transfer_agreement_ref: str | None = Field(default=None, max_length=500)
@@ -70,6 +72,7 @@ class DatasourceFieldResponse(BaseModel):
     name: str
     data_type: str
     nullable: bool
+    classification: str = "internal"
 
 
 class DatasourceResponse(BaseModel):

src/govoplan_datasources/backend/service.py

@@ -33,7 +33,9 @@ from govoplan_core.core.datasources import (
     DatasourceStageInput,
     DatasourceUnavailableError,
     DatasourceValidationError,
+    DatasourceVisibilityPolicyRequest,
     datasource_origins,
+    datasource_visibility_policy_provider,
 )
 from govoplan_core.db.base import utcnow
 from govoplan_datasources.backend.db.models import (
@@ -62,6 +64,14 @@ from govoplan_datasources.backend.tabular import (
     infer_schema,
     normalize_rows,
 )
+from govoplan_datasources.backend.visibility import (
+    VisibilityPlan,
+    apply_visibility,
+    normalize_visibility_policy,
+    visibility_diagnostics,
+    visibility_plan,
+    visible_schema,
+)
 
 
 CATALOGUE_READ_SCOPE = "datasources:catalogue:read"
@@ -335,7 +345,15 @@ class SqlDatasourceProvider:
                         DatasourceGovernanceReferenceRecord.reference == cleaned,
                     )
                 )
-        return tuple(_datasource_dto(item) for item in db.scalars(statement))
+        result: list[DatasourceDescriptor] = []
+        for item in db.scalars(statement):
+            if has_scope(api_principal, ADMIN_SCOPE) or self._can_discover(
+                db,
+                api_principal,
+                item=item,
+            ):
+                result.append(_datasource_dto(item))
+        return tuple(result)
 
     def get_datasource(
         self,
@@ -350,7 +368,15 @@ class SqlDatasourceProvider:
             tenant_id=api_principal.tenant_id,
             datasource_ref=datasource_ref,
         )
-        return _datasource_dto(item) if item is not None else None
+        if item is None:
+            return None
+        if not has_scope(api_principal, ADMIN_SCOPE) and not self._can_discover(
+            db,
+            api_principal,
+            item=item,
+        ):
+            return None
+        return _datasource_dto(item)
 
     def read_datasource(
         self,
@@ -374,41 +400,67 @@ class SqlDatasourceProvider:
             and not request.materialization_ref
             and request.consistency == "current"
         )
-        if read_live:
-            if not item.provider_ref:
-                raise DatasourceUnavailableError(
-                    "This datasource has no live origin."
+        try:
+            if read_live:
+                if not item.provider_ref:
+                    raise DatasourceUnavailableError(
+                        "This datasource has no live origin."
+                    )
+                result = self._read_live(
+                    db,
+                    api_principal,
+                    item=item,
+                    request=request,
+                    limit=limit,
+                    offset=offset,
+                    columns=columns,
                 )
-            result = self._read_live(
+            else:
+                result = self._read_materialized(
+                    db,
+                    api_principal,
+                    item=item,
+                    request=request,
+                    limit=limit,
+                    offset=offset,
+                    columns=columns,
+                )
+        except DatasourceAccessError:
+            _audit_visibility_read(
                 db,
                 api_principal,
                 item=item,
-                request=request,
-                limit=limit,
-                offset=offset,
-                columns=columns,
+                outcome="denied",
+                consistency=request.consistency,
             )
-            if (
-                request.expected_fingerprint
-                and request.expected_fingerprint != result.datasource.fingerprint
-            ):
-                raise DatasourceValidationError(
-                    "The live datasource fingerprint changed; refresh the consuming definition."
-                )
-            return result
-
-        return self._read_materialized(
-            db,
-            item=item,
-            request=request,
-            limit=limit,
-            offset=offset,
-            columns=columns,
+            raise
+        if (
+            request.expected_fingerprint
+            and request.expected_fingerprint != result.datasource.fingerprint
+        ):
+            raise DatasourceValidationError(
+                "The datasource permitted-view fingerprint changed; refresh the consuming definition."
+            )
+        visibility_evidence = tuple(
+            item
+            for item in result.diagnostics
+            if item.code.startswith("datasource.visibility_")
         )
+        if visibility_evidence:
+            _audit_visibility_read(
+                db,
+                api_principal,
+                item=item,
+                outcome="filtered",
+                consistency=request.consistency,
+                diagnostics=visibility_evidence,
+            )
+        return result
 
     def _read_materialized(
         self,
         session: Session,
+        principal: ApiPrincipal,
         *,
         item: DatasourceRecord,
         request: DatasourceReadRequest,
@@ -428,41 +480,114 @@ class SqlDatasourceProvider:
                 else "This datasource has no materialized state."
             )
             raise DatasourceUnavailableError(message)
-        if (
-            request.expected_fingerprint
-            and request.expected_fingerprint != materialization.fingerprint
-        ):
-            raise DatasourceValidationError(
-                "The datasource fingerprint changed; refresh the consuming definition."
-            )
-        _validate_columns(materialization.schema_, columns)
+        source_schema = _fields(materialization.schema_)
+        snapshot_governance = DatasourceGovernance.from_mapping(
+            materialization.governance_snapshot_
+        )
+        snapshot_policies = (
+            (snapshot_governance.visibility_policy,)
+            if snapshot_governance.visibility_policy
+            and snapshot_governance.visibility_policy
+            != _datasource_governance(item).visibility_policy
+            else ()
+        )
+        plan = self._visibility_plan(
+            session,
+            principal,
+            item=item,
+            request=request,
+            materialized=True,
+            schema=source_schema,
+            additional_policies=snapshot_policies,
+        )
+        permitted_schema = visible_schema(source_schema, plan)
+        _validate_field_names(permitted_schema, columns)
         payload = payload_for_materialization(session, materialization)
-        if payload is None:
-            window = materialization.rows[offset : offset + limit]
-        else:
-            backend = self._payload_backends.require(payload.backend)
-            backend.verify(session, payload)
-            window = backend.read_rows(
-                session,
-                payload,
-                offset=offset,
-                limit=limit,
+        if plan.applied:
+            if payload is None:
+                scanned_rows = tuple(materialization.rows[:MAX_STAGE_ROWS])
+            else:
+                backend = self._payload_backends.require(payload.backend)
+                backend.verify(session, payload)
+                scanned_rows = tuple(
+                    backend.read_rows(
+                        session,
+                        payload,
+                        offset=0,
+                        limit=MAX_STAGE_ROWS,
+                    )
+                )
+            permitted_rows, filtered_rows = apply_visibility(
+                scanned_rows,
+                principal=principal.principal,
+                plan=plan,
+                columns=columns,
             )
-        rows = tuple(_select_columns(row, columns) for row in window)
+            candidate_rows = permitted_rows[offset : offset + limit]
+            rows = _bounded_result_rows(
+                candidate_rows,
+                max_bytes=request.max_bytes,
+            )
+            byte_limited = len(rows) < len(candidate_rows)
+            total_rows = len(permitted_rows)
+            scan_limited = len(scanned_rows) < materialization.row_count
+        else:
+            if payload is None:
+                window = materialization.rows[offset : offset + limit]
+            else:
+                backend = self._payload_backends.require(payload.backend)
+                backend.verify(session, payload)
+                window = backend.read_rows(
+                    session,
+                    payload,
+                    offset=offset,
+                    limit=limit,
+                )
+            rows = tuple(_select_columns(row, columns) for row in window)
+            scanned_rows = rows
+            total_rows = materialization.row_count
+            filtered_rows = 0
+            scan_limited = False
+            byte_limited = False
+        view_fingerprint = plan.view_fingerprint(materialization.fingerprint)
         descriptor = _datasource_dto(item)
+        result_descriptor = replace(
+            descriptor,
+            schema=permitted_schema,
+            schema_version=str(materialization.schema_version),
+            fingerprint=view_fingerprint,
+            row_count=total_rows,
+            byte_count=encoded_size(rows),
+        )
+        materialization_dto = replace(
+            _materialization_dto(materialization),
+            schema=permitted_schema,
+            fingerprint=view_fingerprint,
+            row_count=total_rows,
+            byte_count=encoded_size(rows),
+        )
+        if plan.applied:
+            result_descriptor = _opaque_descriptor(result_descriptor)
+            materialization_dto = _opaque_materialization(materialization_dto)
         return DatasourceReadResult(
-            datasource=replace(
-                descriptor,
-                schema=_fields(materialization.schema_),
-                schema_version=str(materialization.schema_version),
-                fingerprint=materialization.fingerprint,
-                row_count=materialization.row_count,
-                byte_count=materialization.byte_count,
-            ),
+            datasource=result_descriptor,
             rows=rows,
-            total_rows=materialization.row_count,
-            truncated=offset + len(rows) < materialization.row_count,
-            materialization=_materialization_dto(materialization),
+            total_rows=total_rows,
+            truncated=(
+                scan_limited or byte_limited or offset + len(rows) < total_rows
+            ),
+            materialization=materialization_dto,
+            returned_bytes=encoded_size(rows),
+            effective_row_limit=limit,
+            effective_byte_limit=request.max_bytes,
+            effective_timeout_ms=request.timeout_ms,
+            diagnostics=visibility_diagnostics(
+                plan,
+                scanned_rows=len(scanned_rows),
+                permitted_rows=total_rows,
+                filtered_rows=filtered_rows,
+                scan_limited=scan_limited,
+            ),
         )
 
     def list_materializations(
@@ -486,7 +611,46 @@ class SqlDatasourceProvider:
             )
             .order_by(DatasourceMaterializationRecord.revision.desc())
         )
-        return tuple(_materialization_dto(row) for row in db.scalars(statement))
+        result: list[DatasourceMaterialization] = []
+        for row in db.scalars(statement):
+            schema = _fields(row.schema_)
+            snapshot = DatasourceGovernance.from_mapping(row.governance_snapshot_)
+            additional_policies = (
+                (snapshot.visibility_policy,)
+                if snapshot.visibility_policy
+                and snapshot.visibility_policy
+                != _datasource_governance(item).visibility_policy
+                else ()
+            )
+            try:
+                plan = self._visibility_plan(
+                    db,
+                    api_principal,
+                    item=item,
+                    request=DatasourceReadRequest(
+                        datasource_ref=datasource_ref,
+                        materialization_ref=_materialization_ref(row.id),
+                        consistency="frozen",
+                    ),
+                    materialized=True,
+                    schema=schema,
+                    additional_policies=additional_policies,
+                )
+            except DatasourceAccessError:
+                continue
+            dto = _materialization_dto(row)
+            if plan.applied:
+                dto = _opaque_materialization(
+                    replace(
+                        dto,
+                        schema=visible_schema(schema, plan),
+                        fingerprint=plan.view_fingerprint(row.fingerprint),
+                        row_count=None,
+                        byte_count=None,
+                    )
+                )
+            result.append(dto)
+        return tuple(result)
 
     def list_stages(
         self,
@@ -551,6 +715,13 @@ class SqlDatasourceProvider:
         )
         rows = normalize_rows(stage.rows)
         schema = infer_schema(rows)
+        governance = replace(
+            governance,
+            visibility_policy=normalize_visibility_policy(
+                governance.visibility_policy,
+                schema=schema,
+            ),
+        )
         fingerprint = fingerprint_rows(rows, schema)
         validation = validate_stage(
             rows=rows,
@@ -919,6 +1090,26 @@ class SqlDatasourceProvider:
             raise DatasourceUnavailableError(
                 "No connector origin provider is currently available."
             )
+        plan = self._visibility_plan(
+            session,
+            principal,
+            item=item,
+            request=request,
+            materialized=False,
+            schema=_fields(item.schema_),
+        )
+        if plan.applied:
+            return self._read_live_visible(
+                session,
+                principal,
+                provider=provider,
+                item=item,
+                request=request,
+                plan=plan,
+                limit=limit,
+                offset=offset,
+                columns=columns,
+            )
         try:
             result = provider.read_origin(
                 session,
@@ -962,6 +1153,203 @@ class SqlDatasourceProvider:
             diagnostics=result.diagnostics,
         )
 
+    def _read_live_visible(
+        self,
+        session: Session,
+        principal: ApiPrincipal,
+        *,
+        provider: object,
+        item: DatasourceRecord,
+        request: DatasourceReadRequest,
+        plan: VisibilityPlan,
+        limit: int,
+        offset: int,
+        columns: tuple[str, ...],
+    ) -> DatasourceReadResult:
+        rows: list[Mapping[str, object]] = []
+        first_result = None
+        last_result = None
+        base_fingerprint: str | None = None
+        elapsed_ms = 0
+        while len(rows) < MAX_STAGE_ROWS:
+            page_limit = min(MAX_READ_ROWS, MAX_STAGE_ROWS - len(rows))
+            try:
+                page = provider.read_origin(
+                    session,
+                    principal,
+                    request=DatasourceOriginReadRequest(
+                        origin_ref=str(item.provider_ref),
+                        limit=page_limit,
+                        offset=len(rows),
+                        expected_fingerprint=base_fingerprint,
+                        max_bytes=request.max_bytes,
+                        timeout_ms=request.timeout_ms,
+                    ),
+                )
+            except DatasourceError:
+                raise
+            except Exception as exc:
+                raise DatasourceUnavailableError(
+                    f"The datasource origin could not be read: {exc}"
+                ) from exc
+            if first_result is None:
+                first_result = page
+                base_fingerprint = page.origin.fingerprint
+            last_result = page
+            elapsed_ms += page.elapsed_ms
+            rows.extend(page.rows)
+            if not page.truncated or not page.rows:
+                break
+        if first_result is None or last_result is None:
+            raise DatasourceUnavailableError(
+                "The datasource origin returned no result."
+            )
+        source_schema = first_result.origin.schema
+        plan = self._visibility_plan(
+            session,
+            principal,
+            item=item,
+            request=request,
+            materialized=False,
+            schema=source_schema,
+        )
+        permitted_schema = visible_schema(source_schema, plan)
+        _validate_field_names(permitted_schema, columns)
+        permitted, filtered_rows = apply_visibility(
+            rows,
+            principal=principal.principal,
+            plan=plan,
+            columns=columns,
+        )
+        window = permitted[offset : offset + limit]
+        window = _bounded_result_rows(window, max_bytes=request.max_bytes)
+        scan_limited = bool(
+            last_result.truncated
+            and (len(rows) >= MAX_STAGE_ROWS or not last_result.rows)
+        )
+        total_rows = len(permitted)
+        view_fingerprint = plan.view_fingerprint(first_result.origin.fingerprint)
+        descriptor = replace(
+            _datasource_dto(item),
+            schema=permitted_schema,
+            schema_version=first_result.origin.schema_version,
+            fingerprint=view_fingerprint,
+            row_count=total_rows,
+            byte_count=encoded_size(window),
+            updated_at=first_result.origin.updated_at,
+        )
+        return DatasourceReadResult(
+            datasource=_opaque_descriptor(descriptor),
+            rows=window,
+            total_rows=total_rows,
+            truncated=(
+                scan_limited
+                or offset + len(window) < total_rows
+                or len(window) < min(limit, max(0, total_rows - offset))
+            ),
+            returned_bytes=encoded_size(window),
+            elapsed_ms=elapsed_ms,
+            effective_row_limit=limit,
+            effective_byte_limit=request.max_bytes,
+            effective_timeout_ms=request.timeout_ms,
+            diagnostics=(
+                *first_result.diagnostics,
+                *visibility_diagnostics(
+                    plan,
+                    scanned_rows=len(rows),
+                    permitted_rows=total_rows,
+                    filtered_rows=filtered_rows,
+                    scan_limited=scan_limited,
+                ),
+            ),
+        )
+
+    def _can_discover(
+        self,
+        session: Session,
+        principal: ApiPrincipal,
+        *,
+        item: DatasourceRecord,
+    ) -> bool:
+        try:
+            self._visibility_plan(
+                session,
+                principal,
+                item=item,
+                request=DatasourceReadRequest(datasource_ref=_datasource_ref(item.id)),
+                materialized=False,
+                schema=_fields(item.schema_),
+                action="discover",
+            )
+        except (DatasourceAccessError, DatasourceValidationError):
+            return False
+        return True
+
+    def _visibility_plan(
+        self,
+        session: Session,
+        principal: ApiPrincipal,
+        *,
+        item: DatasourceRecord,
+        request: DatasourceReadRequest,
+        materialized: bool,
+        schema: Sequence[DatasourceField],
+        action: str = "read",
+        additional_policies: Sequence[Mapping[str, object]] = (),
+    ) -> VisibilityPlan:
+        governance = _datasource_governance(item)
+        policies: list[Mapping[str, object]] = []
+        if governance.visibility_policy:
+            policies.append(governance.visibility_policy)
+        policies.extend(additional_policies)
+        decision_refs: list[str] = []
+        policy_provider = datasource_visibility_policy_provider(self._registry)
+        if policy_provider is None:
+            if governance.access_policy_ref:
+                raise DatasourceAccessError(
+                    "The configured datasource access policy provider is unavailable."
+                )
+        else:
+            try:
+                decision = policy_provider.decide_datasource_visibility(
+                    session,
+                    request=DatasourceVisibilityPolicyRequest(
+                        tenant_id=principal.tenant_id,
+                        datasource_ref=_datasource_ref(item.id),
+                        principal=principal.principal,
+                        action=cast(Any, action),
+                        classification=governance.classification,
+                        policy_ref=governance.access_policy_ref,
+                        consistency=request.consistency,
+                        materialization_ref=request.materialization_ref,
+                    ),
+                )
+            except DatasourceError:
+                raise
+            except Exception as exc:
+                raise DatasourceAccessError(
+                    "Datasource policy evaluation is unavailable."
+                ) from exc
+            if not decision.allowed:
+                raise DatasourceAccessError(
+                    decision.reason or "Datasource policy denied access."
+                )
+            policies.extend(decision.policies)
+            if decision.decision_ref:
+                decision_refs.append(decision.decision_ref)
+        try:
+            return visibility_plan(
+                principal=principal.principal,
+                policies=policies,
+                materialized=materialized,
+                schema=schema,
+                decision_refs=decision_refs,
+            )
+        except DatasourceValidationError as exc:
+            raise DatasourceAccessError(
+                "Datasource visibility policy is invalid and access was denied."
+            ) from exc
+
     def _required_origin(
         self,
         session: Session,
@@ -1422,6 +1810,8 @@ def _datasource_governance(item: DatasourceRecord) -> DatasourceGovernance:
             "classification": item.classification,
             "privacy_profile_ref": item.privacy_profile_ref,
             "retention_policy_ref": item.retention_policy_ref,
+            "access_policy_ref": item.access_policy_ref,
+            "visibility_policy": item.visibility_policy,
             "hold_refs": item.hold_refs,
             "publication_state": item.publication_state,
             "transfer_agreement_ref": item.transfer_agreement_ref,
@@ -1453,6 +1843,13 @@ def _apply_datasource_governance(
     item.classification = governance.classification
     item.privacy_profile_ref = governance.privacy_profile_ref
     item.retention_policy_ref = governance.retention_policy_ref
+    item.access_policy_ref = governance.access_policy_ref
+    item.visibility_policy = dict(
+        normalize_visibility_policy(
+            governance.visibility_policy,
+            schema=_fields(item.schema_),
+        )
+    )
     item.hold_refs = list(governance.hold_refs)
     item.publication_state = governance.publication_state
     item.transfer_agreement_ref = governance.transfer_agreement_ref
@@ -1569,6 +1966,7 @@ def _fields(payload: Sequence[Mapping[str, object]]) -> tuple[DatasourceField, .
             name=str(item.get("name", "")),
             data_type=str(item.get("data_type", "unknown")),
             nullable=bool(item.get("nullable", True)),
+            classification=str(item.get("classification") or "internal"),
         )
         for item in payload
     )
@@ -1617,6 +2015,109 @@ def _validate_columns(
         )
 
 
+def _validate_field_names(
+    schema: Sequence[DatasourceField],
+    columns: tuple[str, ...],
+) -> None:
+    _validate_columns(
+        tuple({"name": field.name} for field in schema),
+        columns,
+    )
+
+
+def _bounded_result_rows(
+    rows: Sequence[Mapping[str, object]],
+    *,
+    max_bytes: int,
+) -> tuple[Mapping[str, object], ...]:
+    result: list[Mapping[str, object]] = []
+    byte_limit = max(1, int(max_bytes))
+    for row in rows:
+        candidate = (*result, row)
+        if encoded_size(candidate) > byte_limit:
+            break
+        result.append(row)
+    return tuple(result)
+
+
+def _opaque_descriptor(item: DatasourceDescriptor) -> DatasourceDescriptor:
+    return replace(
+        item,
+        provider=None,
+        provider_ref=None,
+        provenance={"visibility": "policy_filtered"},
+        metadata={},
+        governance=DatasourceGovernance(
+            classification=item.governance.classification,
+            publication_state=item.governance.publication_state,
+        ),
+    )
+
+
+def _opaque_materialization(
+    item: DatasourceMaterialization,
+) -> DatasourceMaterialization:
+    return replace(
+        item,
+        provenance={"visibility": "policy_filtered"},
+        metadata={},
+        governance=DatasourceGovernance(
+            classification=item.governance.classification,
+            publication_state=item.governance.publication_state,
+        ),
+    )
+
+
+def _audit_visibility_read(
+    session: Session,
+    principal: ApiPrincipal,
+    *,
+    item: DatasourceRecord,
+    outcome: str,
+    consistency: str,
+    diagnostics: Sequence[object] = (),
+) -> None:
+    visibility_summary: dict[str, object] = {}
+    for diagnostic in diagnostics:
+        if getattr(diagnostic, "code", None) != "datasource.visibility_applied":
+            continue
+        details = getattr(diagnostic, "details", {})
+        if isinstance(details, Mapping):
+            visibility_summary = {
+                key: details[key]
+                for key in (
+                    "policy_count",
+                    "field_rule_count",
+                    "row_filter_count",
+                    "scanned_rows",
+                    "permitted_rows",
+                    "filtered_rows",
+                )
+                if key in details
+            }
+    audit_event(
+        session,
+        tenant_id=principal.tenant_id,
+        user_id=getattr(principal.user, "id", None) or principal.account_id,
+        api_key_id=principal.api_key_id,
+        action=f"datasource.read.{outcome}",
+        object_type="datasource",
+        object_id=item.id,
+        details={
+            "outcome": outcome,
+            "consistency": consistency,
+            "policy_ref_configured": bool(item.access_policy_ref),
+            "local_policy_configured": bool(item.visibility_policy),
+            "diagnostic_codes": [
+                str(getattr(diagnostic, "code", ""))
+                for diagnostic in diagnostics
+                if getattr(diagnostic, "code", None)
+            ],
+            "visibility_summary": visibility_summary,
+        },
+    )
+
+
 def _select_columns(
     row: Mapping[str, object],
     columns: tuple[str, ...],

src/govoplan_datasources/backend/tabular.py

@@ -145,6 +145,7 @@ def field_payload(field: DatasourceField) -> dict[str, object]:
         "name": field.name,
         "data_type": field.data_type,
         "nullable": field.nullable,
+        "classification": field.classification,
     }
 
 

src/govoplan_datasources/backend/visibility.py

@@ -0,0 +1,490 @@
+from __future__ import annotations
+
+import hashlib
+import json
+from collections.abc import Mapping, Sequence
+from dataclasses import dataclass
+from typing import cast
+
+from govoplan_core.core.access import PrincipalRef
+from govoplan_core.core.datasources import (
+    DatasourceAccessError,
+    DatasourceField,
+    DatasourceValidationError,
+)
+from govoplan_core.core.tabular_sources import TabularPreviewDiagnostic
+
+
+_ACL_KEYS = frozenset(
+    {
+        "account_ids",
+        "membership_ids",
+        "identity_ids",
+        "group_ids",
+        "role_ids",
+        "service_account_ids",
+        "auth_methods",
+    }
+)
+_CLAIM_KEYS = frozenset(
+    {
+        "account_id",
+        "membership_id",
+        "identity_id",
+        "service_account_id",
+        "group_ids",
+        "role_ids",
+        "function_assignment_ids",
+    }
+)
+_POLICY_KEYS = frozenset({"source_acl", "materialization_acl", "fields", "row_filters"})
+
+
+@dataclass(frozen=True, slots=True)
+class FieldRule:
+    name: str
+    classification: str
+    action: str
+    allowed: bool
+
+
+@dataclass(frozen=True, slots=True)
+class RowFilter:
+    field: str
+    claim: str
+    operator: str
+    allow_null: bool
+
+
+@dataclass(frozen=True, slots=True)
+class VisibilityPlan:
+    applied: bool
+    policy_hashes: tuple[str, ...]
+    field_rules: tuple[FieldRule, ...]
+    row_filters: tuple[RowFilter, ...]
+    principal_fingerprint: str
+    decision_refs: tuple[str, ...] = ()
+
+    def view_fingerprint(self, base_fingerprint: str) -> str:
+        if not self.applied:
+            return base_fingerprint
+        return _digest(
+            {
+                "base_fingerprint": base_fingerprint,
+                "policy_hashes": self.policy_hashes,
+                "principal_fingerprint": self.principal_fingerprint,
+                "decision_refs": self.decision_refs,
+            }
+        )
+
+
+def visibility_plan(
+    *,
+    principal: PrincipalRef,
+    policies: Sequence[Mapping[str, object]],
+    materialized: bool,
+    schema: Sequence[DatasourceField],
+    decision_refs: Sequence[str] = (),
+) -> VisibilityPlan:
+    normalized = tuple(_normalize_policy(policy) for policy in policies if policy)
+    if not normalized:
+        return VisibilityPlan(
+            applied=False,
+            policy_hashes=(),
+            field_rules=(),
+            row_filters=(),
+            principal_fingerprint="",
+        )
+
+    schema_names = {field.name for field in schema}
+    field_rules: list[FieldRule] = []
+    row_filters: list[RowFilter] = []
+    for policy in normalized:
+        source_acl = policy.get("source_acl")
+        if isinstance(source_acl, Mapping) and not _acl_allows(source_acl, principal):
+            raise DatasourceAccessError("Datasource visibility policy denied access.")
+        materialization_acl = policy.get("materialization_acl")
+        if (
+            materialized
+            and isinstance(materialization_acl, Mapping)
+            and not _acl_allows(materialization_acl, principal)
+        ):
+            raise DatasourceAccessError(
+                "Datasource materialization policy denied access."
+            )
+        fields = policy.get("fields", {})
+        if isinstance(fields, Mapping):
+            for name, raw_rule in fields.items():
+                if name not in schema_names:
+                    raise DatasourceValidationError(
+                        f"Visibility policy references unknown field {name!r}."
+                    )
+                assert isinstance(raw_rule, Mapping)
+                field_rules.append(
+                    FieldRule(
+                        name=name,
+                        classification=str(raw_rule["classification"]),
+                        action=str(raw_rule["action"]),
+                        allowed=_acl_allows(raw_rule["allow"], principal),
+                    )
+                )
+        for raw_filter in policy.get("row_filters", ()):
+            assert isinstance(raw_filter, Mapping)
+            field = str(raw_filter["field"])
+            if field not in schema_names:
+                raise DatasourceValidationError(
+                    f"Visibility policy references unknown row-filter field {field!r}."
+                )
+            row_filters.append(
+                RowFilter(
+                    field=field,
+                    claim=str(raw_filter["claim"]),
+                    operator=str(raw_filter["operator"]),
+                    allow_null=bool(raw_filter["allow_null"]),
+                )
+            )
+
+    all_facts = _principal_policy_facts(principal)
+    relevant_fact_keys = _relevant_policy_fact_keys(normalized)
+    principal_payload = {key: all_facts[key] for key in sorted(relevant_fact_keys)}
+    return VisibilityPlan(
+        applied=True,
+        policy_hashes=tuple(_digest(policy) for policy in normalized),
+        field_rules=tuple(field_rules),
+        row_filters=tuple(row_filters),
+        principal_fingerprint=_digest(principal_payload),
+        decision_refs=tuple(sorted(set(decision_refs))),
+    )
+
+
+def normalize_visibility_policy(
+    policy: Mapping[str, object],
+    *,
+    schema: Sequence[DatasourceField] = (),
+) -> Mapping[str, object]:
+    normalized = _normalize_policy(policy)
+    if schema:
+        known = {field.name for field in schema}
+        configured = set(cast(Mapping[str, object], normalized.get("fields", {})))
+        configured.update(
+            str(item["field"])
+            for item in cast(
+                Sequence[Mapping[str, object]], normalized.get("row_filters", ())
+            )
+        )
+        unknown = sorted(configured - known)
+        if unknown:
+            raise DatasourceValidationError(
+                f"Visibility policy references unknown fields: {', '.join(unknown)}."
+            )
+    return normalized
+
+
+def visible_schema(
+    schema: Sequence[DatasourceField],
+    plan: VisibilityPlan,
+) -> tuple[DatasourceField, ...]:
+    classifications: dict[str, str] = {}
+    omitted: set[str] = set()
+    for rule in plan.field_rules:
+        classifications[rule.name] = rule.classification
+        if not rule.allowed and rule.action == "omit":
+            omitted.add(rule.name)
+    return tuple(
+        DatasourceField(
+            name=field.name,
+            data_type=field.data_type,
+            nullable=field.nullable,
+            classification=classifications.get(field.name, field.classification),
+        )
+        for field in schema
+        if field.name not in omitted
+    )
+
+
+def apply_visibility(
+    rows: Sequence[Mapping[str, object]],
+    *,
+    principal: PrincipalRef,
+    plan: VisibilityPlan,
+    columns: Sequence[str] = (),
+) -> tuple[tuple[Mapping[str, object], ...], int]:
+    omitted = {
+        rule.name
+        for rule in plan.field_rules
+        if not rule.allowed and rule.action == "omit"
+    }
+    redacted = {
+        rule.name
+        for rule in plan.field_rules
+        if not rule.allowed and rule.action == "redact"
+    }
+    requested = tuple(dict.fromkeys(columns))
+    result: list[Mapping[str, object]] = []
+    filtered = 0
+    for source_row in rows:
+        if not all(
+            _row_filter_allows(source_row, item, principal) for item in plan.row_filters
+        ):
+            filtered += 1
+            continue
+        names = requested or tuple(source_row)
+        result.append(
+            {
+                name: None if name in redacted else source_row.get(name)
+                for name in names
+                if name not in omitted
+            }
+        )
+    return tuple(result), filtered
+
+
+def visibility_diagnostics(
+    plan: VisibilityPlan,
+    *,
+    scanned_rows: int,
+    permitted_rows: int,
+    filtered_rows: int,
+    scan_limited: bool,
+) -> tuple[TabularPreviewDiagnostic, ...]:
+    if not plan.applied:
+        return ()
+    diagnostics = [
+        TabularPreviewDiagnostic(
+            severity="info",
+            code="datasource.visibility_applied",
+            message="Datasource visibility policy was applied before rows left the provider.",
+            details={
+                "policy_count": len(plan.policy_hashes),
+                "field_rule_count": len(plan.field_rules),
+                "row_filter_count": len(plan.row_filters),
+                "scanned_rows": scanned_rows,
+                "permitted_rows": permitted_rows,
+                "filtered_rows": filtered_rows,
+            },
+        )
+    ]
+    if scan_limited:
+        diagnostics.append(
+            TabularPreviewDiagnostic(
+                severity="warning",
+                code="datasource.visibility_scan_limited",
+                message="The bounded visibility scan ended before the origin was exhausted.",
+                details={"scanned_rows": scanned_rows},
+            )
+        )
+    return tuple(diagnostics)
+
+
+def _normalize_policy(policy: Mapping[str, object]) -> dict[str, object]:
+    unknown = set(policy) - _POLICY_KEYS
+    if unknown:
+        raise DatasourceValidationError(
+            f"Unsupported visibility policy keys: {', '.join(sorted(unknown))}."
+        )
+    result: dict[str, object] = {}
+    for name in ("source_acl", "materialization_acl"):
+        if name in policy:
+            result[name] = _normalize_acl(policy[name], path=name)
+
+    raw_fields = policy.get("fields", {})
+    if not isinstance(raw_fields, Mapping) or len(raw_fields) > 500:
+        raise DatasourceValidationError(
+            "Visibility policy fields must be a mapping of at most 500 fields."
+        )
+    fields: dict[str, object] = {}
+    for raw_name, raw_rule in sorted(raw_fields.items(), key=lambda item: str(item[0])):
+        name = str(raw_name).strip()
+        if not name or not isinstance(raw_rule, Mapping):
+            raise DatasourceValidationError(
+                "Every visibility field rule needs a field name and mapping."
+            )
+        unknown_rule = set(raw_rule) - {"classification", "action", "allow"}
+        if unknown_rule:
+            raise DatasourceValidationError(
+                f"Unsupported visibility field keys for {name!r}: {', '.join(sorted(unknown_rule))}."
+            )
+        classification = str(raw_rule.get("classification") or "restricted").strip()
+        if not classification or len(classification) > 80:
+            raise DatasourceValidationError(
+                "Field classifications must contain 1 to 80 characters."
+            )
+        action = str(raw_rule.get("action") or "redact").strip()
+        if action not in {"redact", "omit"}:
+            raise DatasourceValidationError(
+                "Field visibility action must be redact or omit."
+            )
+        if "allow" not in raw_rule:
+            raise DatasourceValidationError(
+                f"Visibility field rule {name!r} needs an allow ACL."
+            )
+        fields[name] = {
+            "classification": classification,
+            "action": action,
+            "allow": _normalize_acl(raw_rule["allow"], path=f"fields.{name}.allow"),
+        }
+    if fields:
+        result["fields"] = fields
+
+    raw_filters = policy.get("row_filters", ())
+    if not isinstance(raw_filters, Sequence) or isinstance(raw_filters, (str, bytes)):
+        raise DatasourceValidationError("Visibility row_filters must be a list.")
+    if len(raw_filters) > 50:
+        raise DatasourceValidationError(
+            "Visibility policy supports at most 50 row filters."
+        )
+    filters: list[dict[str, object]] = []
+    for raw_filter in raw_filters:
+        if not isinstance(raw_filter, Mapping):
+            raise DatasourceValidationError(
+                "Every visibility row filter must be a mapping."
+            )
+        unknown_filter = set(raw_filter) - {"field", "claim", "operator", "allow_null"}
+        if unknown_filter:
+            raise DatasourceValidationError(
+                f"Unsupported row-filter keys: {', '.join(sorted(unknown_filter))}."
+            )
+        field = str(raw_filter.get("field") or "").strip()
+        claim = str(raw_filter.get("claim") or "").strip()
+        operator = str(raw_filter.get("operator") or "equals").strip()
+        if not field:
+            raise DatasourceValidationError(
+                "Every visibility row filter needs a field."
+            )
+        if claim not in _CLAIM_KEYS:
+            raise DatasourceValidationError(
+                f"Unsupported visibility row-filter claim {claim!r}."
+            )
+        if operator not in {"equals", "in"}:
+            raise DatasourceValidationError("Row-filter operator must be equals or in.")
+        filters.append(
+            {
+                "field": field,
+                "claim": claim,
+                "operator": operator,
+                "allow_null": bool(raw_filter.get("allow_null", False)),
+            }
+        )
+    if filters:
+        result["row_filters"] = filters
+    return result
+
+
+def _normalize_acl(value: object, *, path: str) -> dict[str, list[str]]:
+    if not isinstance(value, Mapping):
+        raise DatasourceValidationError(
+            f"Visibility policy {path} must be an ACL mapping."
+        )
+    unknown = set(value) - _ACL_KEYS
+    if unknown:
+        raise DatasourceValidationError(
+            f"Unsupported ACL selectors in {path}: {', '.join(sorted(unknown))}."
+        )
+    result: dict[str, list[str]] = {}
+    for key in sorted(value):
+        raw_values = value[key]
+        if not isinstance(raw_values, Sequence) or isinstance(raw_values, (str, bytes)):
+            raise DatasourceValidationError(
+                f"Visibility ACL selector {path}.{key} must be a list."
+            )
+        if len(raw_values) > 250:
+            raise DatasourceValidationError(
+                f"Visibility ACL selector {path}.{key} is limited to 250 entries."
+            )
+        normalized = sorted(
+            {str(item).strip() for item in raw_values if str(item).strip()}
+        )
+        result[key] = normalized
+    return result
+
+
+def _acl_allows(acl: Mapping[str, object], principal: PrincipalRef) -> bool:
+    facts = _principal_policy_facts(principal)
+    return any(set(values) & set(facts.get(key, ())) for key, values in acl.items())
+
+
+def _principal_policy_facts(principal: PrincipalRef) -> dict[str, tuple[str, ...]]:
+    return {
+        "account_ids": _optional_tuple(principal.account_id),
+        "membership_ids": _optional_tuple(principal.membership_id),
+        "identity_ids": _optional_tuple(principal.identity_id),
+        "group_ids": tuple(sorted(principal.group_ids)),
+        "role_ids": tuple(sorted(principal.role_ids)),
+        "service_account_ids": _optional_tuple(principal.service_account_id),
+        "auth_methods": (principal.auth_method,),
+        "function_assignment_ids": tuple(sorted(principal.function_assignment_ids)),
+    }
+
+
+def _claim_values(principal: PrincipalRef, claim: str) -> tuple[str, ...]:
+    facts = _principal_policy_facts(principal)
+    aliases = {
+        "account_id": "account_ids",
+        "membership_id": "membership_ids",
+        "identity_id": "identity_ids",
+        "service_account_id": "service_account_ids",
+    }
+    return facts.get(aliases.get(claim, claim), ())
+
+
+def _relevant_policy_fact_keys(
+    policies: Sequence[Mapping[str, object]],
+) -> set[str]:
+    keys: set[str] = set()
+    aliases = {
+        "account_id": "account_ids",
+        "membership_id": "membership_ids",
+        "identity_id": "identity_ids",
+        "service_account_id": "service_account_ids",
+    }
+    for policy in policies:
+        for acl_name in ("source_acl", "materialization_acl"):
+            acl = policy.get(acl_name)
+            if isinstance(acl, Mapping):
+                keys.update(str(key) for key in acl)
+        fields = policy.get("fields")
+        if isinstance(fields, Mapping):
+            for rule in fields.values():
+                if isinstance(rule, Mapping) and isinstance(rule.get("allow"), Mapping):
+                    keys.update(str(key) for key in rule["allow"])
+        for row_filter in policy.get("row_filters", ()):
+            if isinstance(row_filter, Mapping):
+                claim = str(row_filter.get("claim") or "")
+                keys.add(aliases.get(claim, claim))
+    return keys
+
+
+def _row_filter_allows(
+    row: Mapping[str, object],
+    item: RowFilter,
+    principal: PrincipalRef,
+) -> bool:
+    value = row.get(item.field)
+    if value is None:
+        return item.allow_null
+    claims = set(_claim_values(principal, item.claim))
+    if not claims:
+        return False
+    if isinstance(value, Sequence) and not isinstance(value, (str, bytes)):
+        values = {str(candidate) for candidate in value}
+        return bool(values & claims)
+    return str(value) in claims
+
+
+def _optional_tuple(value: object | None) -> tuple[str, ...]:
+    return (str(value),) if value is not None and str(value) else ()
+
+
+def _digest(value: object) -> str:
+    encoded = json.dumps(value, sort_keys=True, separators=(",", ":"), default=str)
+    return hashlib.sha256(encoded.encode("utf-8")).hexdigest()
+
+
+__all__ = [
+    "VisibilityPlan",
+    "apply_visibility",
+    "normalize_visibility_policy",
+    "visibility_diagnostics",
+    "visibility_plan",
+    "visible_schema",
+]