Declare policy View surfaces

README.md

@@ -1,5 +1,9 @@
 # GovOPlaN Policy
 
+<!-- govoplan-repository-type:start -->
+**Repository type:** module (platform).
+<!-- govoplan-repository-type:end -->
+
 `govoplan-policy` owns policy and retention API route contributions and the
 retention administration WebUI sections during the GovOPlaN module split.
 
@@ -16,3 +20,19 @@ Hierarchical policy evaluation, delegation ceilings, and write simulations are
 implemented in `govoplan_policy.backend.hierarchy`. Privacy retention uses that
 shared helper and exposes `/api/v1/admin/privacy-retention/policies/{scope}/simulate`
 for preflight checks before saving lower-level policy changes.
+
+The module also provides the optional
+`policy.schedulingParticipantPrivacy` capability. Scheduling owns each
+request's participant-visibility setting; Policy can only preserve or narrow
+it. The resolver reads an optional `maximum_visibility` ceiling from the
+`scheduling_participant_privacy_policy` object in system and tenant settings.
+Missing policy is unrestricted, while malformed explicit policy fails closed
+to aggregate-only visibility. This resolver slice intentionally has no policy
+management endpoint or UI yet.
+
+Policy also provides `policy.definitionGovernance` for Dataflow and Workflow
+libraries. It evaluates view, edit, run/start, reuse, derive, and automation
+actions across system, tenant, group, and user scopes. Templates cannot run or
+be automated. Derived definitions retain ancestor ceilings, and every
+decision includes the ordered Policy source path and effective limits so a UI
+can explain why an action is available or blocked.

docs/POLICY_DECISION_PROVENANCE.md

@@ -20,6 +20,34 @@ When retention needs audit-log storage behavior, it requests the
 `audit.retention` capability; it does not import audit module tables or
 providers directly.
 
+## Scheduling Participant Privacy
+
+Policy exposes `policy.schedulingParticipantPrivacy` as an optional restriction
+hook. Scheduling supplies the request-level visibility choice and remains
+responsible for its secure fallback when Policy is absent. The provider never
+broadens that choice.
+
+System and tenant settings may contain:
+
+```json
+{
+  "scheduling_participant_privacy_policy": {
+    "maximum_visibility": "aggregates_only"
+  }
+}
+```
+
+`maximum_visibility` is either `aggregates_only` or `names_and_statuses`.
+Omitted settings impose no additional restriction. Effective visibility is the
+most restrictive of the Scheduling request, the system ceiling, and the tenant
+ceiling. Explicit malformed policy fails closed to `aggregates_only` and is
+reported in decision details without echoing the invalid stored value. Policy
+sources include only explicitly configured or invalid system and tenant steps.
+
+The current slice is resolver-only. A managed write API and administration UI
+must add validation, audit, parent-ceiling enforcement, and configuration
+safety registration before operators can edit this setting through GovOPlaN.
+
 ## Backend DTOs
 
 Use `govoplan_core.core.policy.PolicyDecision` for explainable policy results:

package.json

@@ -1,6 +1,6 @@
 {
   "name": "@govoplan/policy-webui",
-  "version": "0.1.7",
+  "version": "0.1.9",
   "private": true,
   "type": "module",
   "main": "webui/src/index.ts",
@@ -18,7 +18,7 @@
     "LICENSE"
   ],
   "peerDependencies": {
-    "@govoplan/core-webui": "^0.1.7",
+    "@govoplan/core-webui": "^0.1.9",
     "lucide-react": "^1.23.0",
     "react": "^19.0.0",
     "react-dom": "^19.0.0",

pyproject.toml

@@ -4,13 +4,13 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "govoplan-policy"
-version = "0.1.7"
+version = "0.1.9"
 description = "GovOPlaN policy platform module."
 readme = "README.md"
 requires-python = ">=3.12"
 authors = [{ name = "GovOPlaN" }]
 dependencies = [
-  "govoplan-core>=0.1.7",
+  "govoplan-core>=0.1.9",
 ]
 
 [tool.setuptools.packages.find]

src/govoplan_policy/backend/api/v1/routes.py

@@ -14,9 +14,9 @@ from govoplan_core.core.configuration_control import (
 )
 from govoplan_core.core.policy import PolicyDecision, PolicySourceStep
 from govoplan_core.db.session import get_session
+from govoplan_core.privacy.schemas import RETENTION_POLICY_FIELD_KEYS
 from govoplan_policy.backend.retention import (
     PrivacyPolicyError,
-    RETENTION_POLICY_FIELD_KEYS,
     apply_retention_policy,
     effective_privacy_policy,
     effective_privacy_policy_sources,

src/govoplan_policy/backend/api/v1/schemas.py

@@ -1,43 +1,10 @@
 from __future__ import annotations
 
-from datetime import datetime
 from typing import Any, Literal
 
-from pydantic import BaseModel, ConfigDict, Field, field_validator
+from pydantic import BaseModel, ConfigDict, Field
 
-
-RETENTION_DAY_KEYS = (
-    "raw_campaign_json_retention_days",
-    "generated_eml_retention_days",
-    "stored_report_detail_retention_days",
-    "mock_mailbox_retention_days",
-    "audit_detail_retention_days",
-)
-
-
-RETENTION_POLICY_FIELD_KEYS = (
-    "store_raw_campaign_json",
-    *RETENTION_DAY_KEYS,
-    "audit_detail_level",
-)
-
-
-def default_allow_lower_level_limits() -> dict[str, bool]:
-    return {key: True for key in RETENTION_POLICY_FIELD_KEYS}
-
-
-def normalize_allow_lower_level_limits(value: Any, *, fill_defaults: bool) -> dict[str, bool] | None:
-    if value in (None, ""):
-        return default_allow_lower_level_limits() if fill_defaults else None
-    if not isinstance(value, dict):
-        raise ValueError("allow_lower_level_limits must be an object")
-    normalized = default_allow_lower_level_limits() if fill_defaults else {}
-    for key, allowed in value.items():
-        clean_key = str(key)
-        if clean_key not in RETENTION_POLICY_FIELD_KEYS:
-            raise ValueError(f"Unknown retention policy field: {clean_key}")
-        normalized[clean_key] = bool(allowed)
-    return normalized
+from govoplan_core.privacy.schemas import PrivacyRetentionPolicyItem, PrivacyRetentionPolicyPatchItem
 
 
 class PolicySourceStepItem(BaseModel):
@@ -49,42 +16,6 @@ class PolicySourceStepItem(BaseModel):
     policy: dict[str, Any] = Field(default_factory=dict)
 
 
-class PrivacyRetentionPolicyItem(BaseModel):
-    model_config = ConfigDict(extra="forbid")
-
-    store_raw_campaign_json: bool = True
-    raw_campaign_json_retention_days: int | None = Field(default=None, ge=0)
-    generated_eml_retention_days: int | None = Field(default=None, ge=0)
-    stored_report_detail_retention_days: int | None = Field(default=None, ge=0)
-    mock_mailbox_retention_days: int | None = Field(default=None, ge=0)
-    audit_detail_retention_days: int | None = Field(default=None, ge=0)
-    audit_detail_level: Literal["full", "redacted", "minimal"] = "full"
-    allow_lower_level_limits: dict[str, bool] = Field(default_factory=default_allow_lower_level_limits)
-
-    @field_validator("allow_lower_level_limits", mode="before")
-    @classmethod
-    def _normalize_allow_lower_level_limits(cls, value: Any) -> Any:
-        return normalize_allow_lower_level_limits(value, fill_defaults=True)
-
-
-class PrivacyRetentionPolicyPatchItem(BaseModel):
-    model_config = ConfigDict(extra="forbid")
-
-    store_raw_campaign_json: bool | None = None
-    raw_campaign_json_retention_days: int | None = Field(default=None, ge=0)
-    generated_eml_retention_days: int | None = Field(default=None, ge=0)
-    stored_report_detail_retention_days: int | None = Field(default=None, ge=0)
-    mock_mailbox_retention_days: int | None = Field(default=None, ge=0)
-    audit_detail_retention_days: int | None = Field(default=None, ge=0)
-    audit_detail_level: Literal["full", "redacted", "minimal"] | None = None
-    allow_lower_level_limits: dict[str, bool] | None = None
-
-    @field_validator("allow_lower_level_limits", mode="before")
-    @classmethod
-    def _normalize_allow_lower_level_limits(cls, value: Any) -> Any:
-        return normalize_allow_lower_level_limits(value, fill_defaults=False)
-
-
 class PrivacyRetentionPolicyScopeRequest(BaseModel):
     model_config = ConfigDict(extra="forbid")
 

src/govoplan_policy/backend/definition_governance.py

@@ -0,0 +1,307 @@
+from __future__ import annotations
+
+from collections.abc import Mapping
+from typing import Any
+
+from govoplan_core.core.policy import (
+    DefinitionGovernanceRequest,
+    PolicyDecision,
+    PolicySourceStep,
+)
+from govoplan_core.security.module_permissions import scopes_grant_compatible
+
+
+class DefinitionGovernancePolicyProvider:
+    """Resolve flow-library actions without importing a domain module."""
+
+    def resolve_definition_action(
+        self,
+        *,
+        request: DefinitionGovernanceRequest,
+    ) -> PolicyDecision:
+        source_path = _source_path(request)
+        effective = _effective_limits(request)
+        visible, visibility_reason = _visible(
+            request,
+            effective=effective,
+        )
+        action = request.action
+
+        if action == "view":
+            return _decision(
+                visible,
+                visibility_reason,
+                source_path=source_path,
+                request=request,
+                effective=effective,
+            )
+
+        if action == "edit":
+            editable, reason = _editable(request)
+            return _decision(
+                editable,
+                reason,
+                source_path=source_path,
+                request=request,
+                effective=effective,
+            )
+
+        if not visible:
+            return _decision(
+                False,
+                visibility_reason,
+                source_path=source_path,
+                request=request,
+                effective=effective,
+            )
+
+        if action == "run":
+            if request.definition_kind == "template":
+                return _decision(
+                    False,
+                    "Templates must be derived into a complete flow before execution.",
+                    source_path=source_path,
+                    request=request,
+                    effective=effective,
+                )
+            if request.status != "active":
+                return _decision(
+                    False,
+                    "Only active flow revisions can be executed.",
+                    source_path=source_path,
+                    request=request,
+                    effective=effective,
+                )
+            return _decision(
+                effective["allow_run"],
+                (
+                    None
+                    if effective["allow_run"]
+                    else "Execution is disabled by definition or ancestor policy."
+                ),
+                source_path=source_path,
+                request=request,
+                effective=effective,
+            )
+
+        if action in {"reuse", "derive"}:
+            return _decision(
+                effective["allow_reuse"],
+                (
+                    None
+                    if effective["allow_reuse"]
+                    else "Reuse is disabled by definition or ancestor policy."
+                ),
+                source_path=source_path,
+                request=request,
+                effective=effective,
+            )
+
+        if action == "automate":
+            if request.definition_kind == "template":
+                return _decision(
+                    False,
+                    "Templates cannot be automated.",
+                    source_path=source_path,
+                    request=request,
+                    effective=effective,
+                )
+            if request.status != "active":
+                return _decision(
+                    False,
+                    "Automation requires an active flow revision.",
+                    source_path=source_path,
+                    request=request,
+                    effective=effective,
+                )
+            allowed = effective["allow_run"] and effective["allow_automation"]
+            return _decision(
+                allowed,
+                (
+                    None
+                    if allowed
+                    else "Automation is disabled by definition or ancestor policy."
+                ),
+                source_path=source_path,
+                request=request,
+                effective=effective,
+            )
+
+        return _decision(
+            False,
+            f"Unsupported definition action: {action}",
+            source_path=source_path,
+            request=request,
+            effective=effective,
+        )
+
+
+def _visible(
+    request: DefinitionGovernanceRequest,
+    *,
+    effective: Mapping[str, bool],
+) -> tuple[bool, str | None]:
+    scope = request.definition_scope
+    actor = request.actor
+    if scope.scope_type == "system":
+        if _has_scope(actor.scopes, "system:governance:read") or _has_scope(
+            actor.scopes,
+            "system:governance:write",
+        ):
+            return True, None
+        if effective["inherit_to_lower_scopes"]:
+            return True, None
+        return False, "The system definition is not inherited by lower scopes."
+    if actor.tenant_id != request.tenant_id:
+        return False, "The definition belongs to another tenant."
+    if scope.scope_type == "tenant":
+        if scope.scope_id != request.tenant_id:
+            return False, "The definition belongs to another tenant."
+        return True, None
+    if scope.scope_type == "group":
+        if scope.scope_id in actor.group_ids:
+            return True, None
+        return False, "The definition is limited to another group."
+    if scope.scope_type == "user":
+        if scope.scope_id in {actor.membership_id, actor.account_id}:
+            return True, None
+        return False, "The definition is limited to another user."
+    return False, "The definition scope is invalid."
+
+
+def _editable(request: DefinitionGovernanceRequest) -> tuple[bool, str | None]:
+    scope = request.definition_scope
+    actor = request.actor
+    if scope.scope_type == "system":
+        allowed = _has_scope(actor.scopes, "system:governance:write")
+        return (
+            allowed,
+            None if allowed else "System definitions require system governance permission.",
+        )
+    if actor.tenant_id != request.tenant_id:
+        return False, "Definitions from another tenant are read-only."
+    if scope.scope_type == "tenant":
+        allowed = scope.scope_id == request.tenant_id
+        return allowed, None if allowed else "Inherited tenant definitions are read-only."
+    if scope.scope_type == "group":
+        allowed = scope.scope_id in actor.group_ids
+        return allowed, None if allowed else "Definitions from another group are read-only."
+    if scope.scope_type == "user":
+        allowed = scope.scope_id in {actor.membership_id, actor.account_id}
+        return allowed, None if allowed else "Definitions from another user are read-only."
+    return False, "The definition scope is invalid."
+
+
+def _effective_limits(
+    request: DefinitionGovernanceRequest,
+) -> dict[str, bool]:
+    ancestor = request.context.get("ancestor_limits")
+    ancestor_limits = ancestor if isinstance(ancestor, Mapping) else {}
+    return {
+        "inherit_to_lower_scopes": request.inherit_to_lower_scopes
+        and _ancestor_flag(ancestor_limits, "inherit_to_lower_scopes"),
+        "allow_run": request.allow_run
+        and _ancestor_flag(ancestor_limits, "allow_run"),
+        "allow_reuse": request.allow_reuse
+        and _ancestor_flag(ancestor_limits, "allow_reuse"),
+        "allow_automation": request.allow_automation
+        and _ancestor_flag(ancestor_limits, "allow_automation"),
+    }
+
+
+def _ancestor_flag(value: Mapping[str, Any], key: str) -> bool:
+    raw = value.get(key)
+    return True if raw is None else raw is True
+
+
+def _source_path(
+    request: DefinitionGovernanceRequest,
+) -> tuple[PolicySourceStep, ...]:
+    steps: list[PolicySourceStep] = []
+    ancestor = request.context.get("ancestor_limits")
+    if isinstance(ancestor, Mapping):
+        source = request.context.get("ancestor_source")
+        source_mapping = source if isinstance(source, Mapping) else {}
+        scope_type = str(source_mapping.get("scope_type") or "system")
+        scope_id = source_mapping.get("scope_id")
+        if scope_type in {"system", "tenant", "group", "user"}:
+            steps.append(
+                PolicySourceStep(
+                    scope_type=scope_type,  # type: ignore[arg-type]
+                    scope_id=str(scope_id) if scope_id is not None else None,
+                    label=str(source_mapping.get("label") or "Ancestor definition"),
+                    applied_fields=tuple(sorted(str(key) for key in ancestor)),
+                    policy=dict(ancestor),
+                )
+            )
+    scope = request.definition_scope
+    steps.append(
+        PolicySourceStep(
+            scope_type=scope.scope_type,
+            scope_id=scope.scope_id,
+            label=f"{scope.scope_type.capitalize()} definition",
+            applied_fields=(
+                "definition_kind",
+                "inherit_to_lower_scopes",
+                "allow_run",
+                "allow_reuse",
+                "allow_automation",
+            ),
+            policy={
+                "definition_kind": request.definition_kind,
+                "status": request.status,
+                "inherit_to_lower_scopes": request.inherit_to_lower_scopes,
+                "allow_run": request.allow_run,
+                "allow_reuse": request.allow_reuse,
+                "allow_automation": request.allow_automation,
+            },
+        )
+    )
+    target = request.target_scope
+    if target.path != scope.path:
+        steps.append(
+            PolicySourceStep(
+                scope_type=target.scope_type,
+                scope_id=target.scope_id,
+                label=f"{target.scope_type.capitalize()} use context",
+                applied_fields=("action",),
+                policy={"action": request.action},
+            )
+        )
+    return tuple(steps)
+
+
+def _decision(
+    allowed: bool,
+    reason: str | None,
+    *,
+    source_path: tuple[PolicySourceStep, ...],
+    request: DefinitionGovernanceRequest,
+    effective: Mapping[str, bool],
+) -> PolicyDecision:
+    return PolicyDecision(
+        allowed=allowed,
+        reason=reason,
+        source_path=source_path,
+        requirements=(
+            ()
+            if allowed
+            else (f"{request.module_id}.definition.{request.action}",)
+        ),
+        details={
+            "module_id": request.module_id,
+            "definition_ref": request.definition_ref,
+            "action": request.action,
+            "definition_scope": request.definition_scope.path,
+            "target_scope": request.target_scope.path,
+            "definition_kind": request.definition_kind,
+            "effective_limits": dict(effective),
+        },
+    )
+
+
+def _has_scope(scopes: object, required: str) -> bool:
+    return scopes_grant_compatible(scopes, required)  # type: ignore[arg-type]
+
+
+__all__ = ["DefinitionGovernancePolicyProvider"]

src/govoplan_policy/backend/hierarchy.py

@@ -66,21 +66,27 @@ def validate_hierarchical_policy_patch(
     relock_message: Callable[[str], str] | None = None,
 ) -> tuple[PolicyValidationIssue, ...]:
     issues: list[PolicyValidationIssue] = []
-    lock_message = locked_field_message or (lambda field: f"{field} is locked by the parent policy.")
-    reenable_message = relock_message or (lambda field: f"{field} limiting cannot be re-enabled below a parent policy lock.")
+    lock_message = locked_field_message or (lambda field_name: f"{field_name} is locked by the parent policy.")
+    reenable_message = relock_message or (
+        lambda field_name: f"{field_name} limiting cannot be re-enabled below a parent policy lock."
+    )
 
-    for field in field_keys:
-        if field in patch and patch.get(field) is not None and not parent_allow_lower_level_limits.get(field, True):
+    for field_name in field_keys:
+        if (
+            field_name in patch
+            and patch.get(field_name) is not None
+            and not parent_allow_lower_level_limits.get(field_name, True)
+        ):
             issues.append(PolicyValidationIssue(
                 code="field_locked_by_parent",
-                field=field,
-                message=lock_message(field),
+                field=field_name,
+                message=lock_message(field_name),
             ))
 
     patch_allow = patch.get("allow_lower_level_limits")
     if isinstance(patch_allow, Mapping):
-        for field, allowed in patch_allow.items():
-            clean_field = str(field)
+        for field_name, allowed in patch_allow.items():
+            clean_field = str(field_name)
             if bool(allowed) and not parent_allow_lower_level_limits.get(clean_field, True):
                 issues.append(PolicyValidationIssue(
                     code="lower_level_limit_locked_by_parent",
@@ -125,9 +131,9 @@ def simulate_hierarchical_policy_change(
         relock_message=relock_message,
     )
     changed_fields = tuple(
-        field
-        for field in (*field_keys, "allow_lower_level_limits")
-        if field in patch and current_policy.get(field) != patch.get(field)
+        field_name
+        for field_name in (*field_keys, "allow_lower_level_limits")
+        if field_name in patch and current_policy.get(field_name) != patch.get(field_name)
     )
     allowed = not any(issue.severity == "blocker" for issue in issues)
     decision = PolicyDecision(

src/govoplan_policy/backend/manifest.py

@@ -1,8 +1,18 @@
 from __future__ import annotations
 
 from govoplan_core.core.access import CAPABILITY_AUTH_PERMISSION_EVALUATOR, CAPABILITY_AUTH_PRINCIPAL_RESOLVER
-from govoplan_core.core.policy import CAPABILITY_POLICY_PRIVACY_RETENTION
-from govoplan_core.core.modules import FrontendModule, ModuleContext, ModuleManifest
+from govoplan_core.core.policy import (
+    CAPABILITY_POLICY_DEFINITION_GOVERNANCE,
+    CAPABILITY_POLICY_PRIVACY_RETENTION,
+    CAPABILITY_POLICY_SCHEDULING_PARTICIPANT_PRIVACY,
+)
+from govoplan_core.core.modules import (
+    FrontendModule,
+    ModuleContext,
+    ModuleInterfaceProvider,
+    ModuleManifest,
+)
+from govoplan_core.core.views import ViewSurface
 
 
 def _route_factory(context: ModuleContext):
@@ -19,18 +29,48 @@ def _privacy_retention_service(context: ModuleContext) -> object:
     return SqlPrivacyRetentionService()
 
 
+def _scheduling_participant_privacy_policy(context: ModuleContext) -> object:
+    del context
+    from govoplan_policy.backend.scheduling_privacy import SqlSchedulingParticipantPrivacyPolicy
+
+    return SqlSchedulingParticipantPrivacyPolicy()
+
+
+def _definition_governance_policy(context: ModuleContext) -> object:
+    del context
+    from govoplan_policy.backend.definition_governance import (
+        DefinitionGovernancePolicyProvider,
+    )
+
+    return DefinitionGovernancePolicyProvider()
+
+
 manifest = ModuleManifest(
     id="policy",
     name="Policy",
-    version="0.1.7",
+    version="0.1.9",
     required_capabilities=(CAPABILITY_AUTH_PRINCIPAL_RESOLVER, CAPABILITY_AUTH_PERMISSION_EVALUATOR),
+    provides_interfaces=(
+        ModuleInterfaceProvider(
+            name="policy.definition_governance",
+            version="0.1.0",
+        ),
+    ),
     route_factory=_route_factory,
     frontend=FrontendModule(
         module_id="policy",
         package_name="@govoplan/policy-webui",
+        view_surfaces=(
+            ViewSurface(id="policy.admin.system-retention", module_id="policy", kind="section", label="System retention", order=80),
+            ViewSurface(id="policy.admin.tenant-retention", module_id="policy", kind="section", label="Tenant retention", order=80),
+            ViewSurface(id="policy.admin.group-retention", module_id="policy", kind="section", label="Group retention", order=80),
+            ViewSurface(id="policy.admin.user-retention", module_id="policy", kind="section", label="User retention", order=80),
+        ),
     ),
     capability_factories={
+        CAPABILITY_POLICY_DEFINITION_GOVERNANCE: _definition_governance_policy,
         CAPABILITY_POLICY_PRIVACY_RETENTION: _privacy_retention_service,
+        CAPABILITY_POLICY_SCHEDULING_PARTICIPANT_PRIVACY: _scheduling_participant_privacy_policy,
     },
 )
 

src/govoplan_policy/backend/retention.py

@@ -3,9 +3,9 @@ from __future__ import annotations
 import json
 from datetime import datetime, timedelta, timezone
 from pathlib import Path
-from typing import Any, Literal
+from typing import Any
 
-from pydantic import BaseModel, ConfigDict, Field, field_validator
+from pydantic import field_validator
 from sqlalchemy.orm import Session
 
 from govoplan_core.admin.settings import get_system_settings
@@ -26,6 +26,14 @@ from govoplan_core.core.access import (
 from govoplan_core.core.policy import PolicySourceStep, policy_source_step
 from govoplan_core.core.runtime import get_registry
 from govoplan_core.admin.models import SystemSettings
+from govoplan_core.privacy.schemas import (
+    RETENTION_DAY_KEYS,
+    RETENTION_POLICY_FIELD_KEYS,
+    PrivacyRetentionPolicyItem,
+    PrivacyRetentionPolicyPatchItem,
+    default_allow_lower_level_limits,
+    normalize_allow_lower_level_limits,
+)
 from govoplan_core.settings import settings
 from govoplan_core.tenancy.scope import Tenant
 from govoplan_policy.backend.hierarchy import (
@@ -35,38 +43,8 @@ from govoplan_policy.backend.hierarchy import (
 )
 
 PRIVACY_POLICY_SETTINGS_KEY = "privacy_retention_policy"
-RETENTION_DAY_KEYS = (
-    "raw_campaign_json_retention_days",
-    "generated_eml_retention_days",
-    "stored_report_detail_retention_days",
-    "mock_mailbox_retention_days",
-    "audit_detail_retention_days",
-)
-RETENTION_POLICY_FIELD_KEYS = (
-    "store_raw_campaign_json",
-    *RETENTION_DAY_KEYS,
-    "audit_detail_level",
-)
 AUDIT_DETAIL_LEVEL_ORDER = {"full": 0, "redacted": 1, "minimal": 2}
 
-
-def default_allow_lower_level_limits() -> dict[str, bool]:
-    return {key: True for key in RETENTION_POLICY_FIELD_KEYS}
-
-
-def _normalize_allow_lower_level_limits(value: Any, *, fill_defaults: bool) -> dict[str, bool] | None:
-    if value in (None, ""):
-        return default_allow_lower_level_limits() if fill_defaults else None
-    if not isinstance(value, dict):
-        raise ValueError("allow_lower_level_limits must be an object")
-    normalized = default_allow_lower_level_limits() if fill_defaults else {}
-    for key, allowed in value.items():
-        clean_key = str(key)
-        if clean_key not in RETENTION_POLICY_FIELD_KEYS:
-            raise ValueError(f"Unknown retention policy field: {clean_key}")
-        normalized[clean_key] = bool(allowed)
-    return normalized
-
 AUDIT_DETAIL_REDACTION_KEYS = {
     "attachments",
     "bcc",
@@ -91,18 +69,7 @@ AUDIT_DETAIL_REDACTION_KEYS = {
 }
 
 
-class PrivacyRetentionPolicy(BaseModel):
-    model_config = ConfigDict(extra="forbid")
-
-    store_raw_campaign_json: bool = True
-    raw_campaign_json_retention_days: int | None = Field(default=None, ge=0)
-    generated_eml_retention_days: int | None = Field(default=None, ge=0)
-    stored_report_detail_retention_days: int | None = Field(default=None, ge=0)
-    mock_mailbox_retention_days: int | None = Field(default=None, ge=0)
-    audit_detail_retention_days: int | None = Field(default=None, ge=0)
-    audit_detail_level: Literal["full", "redacted", "minimal"] = "full"
-    allow_lower_level_limits: dict[str, bool] = Field(default_factory=default_allow_lower_level_limits)
-
+class PrivacyRetentionPolicy(PrivacyRetentionPolicyItem):
     @field_validator(
         "raw_campaign_json_retention_days",
         "generated_eml_retention_days",
@@ -120,21 +87,10 @@ class PrivacyRetentionPolicy(BaseModel):
     @field_validator("allow_lower_level_limits", mode="before")
     @classmethod
     def _normalize_allow_lower_level_limits(cls, value: Any) -> Any:
-        return _normalize_allow_lower_level_limits(value, fill_defaults=True)
+        return normalize_allow_lower_level_limits(value, fill_defaults=True)
 
 
-class PrivacyRetentionPolicyPatch(BaseModel):
-    model_config = ConfigDict(extra="forbid")
-
-    store_raw_campaign_json: bool | None = None
-    raw_campaign_json_retention_days: int | None = Field(default=None, ge=0)
-    generated_eml_retention_days: int | None = Field(default=None, ge=0)
-    stored_report_detail_retention_days: int | None = Field(default=None, ge=0)
-    mock_mailbox_retention_days: int | None = Field(default=None, ge=0)
-    audit_detail_retention_days: int | None = Field(default=None, ge=0)
-    audit_detail_level: Literal["full", "redacted", "minimal"] | None = None
-    allow_lower_level_limits: dict[str, bool] | None = None
-
+class PrivacyRetentionPolicyPatch(PrivacyRetentionPolicyPatchItem):
     @field_validator(*RETENTION_DAY_KEYS, mode="before")
     @classmethod
     def _blank_string_is_none(cls, value: Any) -> Any:
@@ -145,7 +101,7 @@ class PrivacyRetentionPolicyPatch(BaseModel):
     @field_validator("allow_lower_level_limits", mode="before")
     @classmethod
     def _normalize_allow_lower_level_limits(cls, value: Any) -> Any:
-        return _normalize_allow_lower_level_limits(value, fill_defaults=False)
+        return normalize_allow_lower_level_limits(value, fill_defaults=False)
 
 
 class PrivacyPolicyError(RuntimeError):
@@ -524,49 +480,130 @@ def set_privacy_policy_for_scope(
 ) -> dict[str, Any]:
     clean_scope = scope_type.strip().casefold()
     if clean_scope == "system":
-        item = get_system_settings(session)
-        validated = set_privacy_policy(item, PrivacyRetentionPolicy.model_validate(policy or {}))
-        session.add(item)
-        return validated.model_dump(mode="json")
-    patch = PrivacyRetentionPolicyPatch.model_validate(policy or {}).model_dump(mode="json", exclude_none=True)
-    _validate_privacy_patch_against_parent(parent_privacy_policy(session, tenant_id=tenant_id, scope_type=clean_scope, scope_id=scope_id or (tenant_id if clean_scope == "tenant" else None)), patch)
-    if clean_scope == "tenant":
-        tenant = session.get(Tenant, scope_id or tenant_id)
-        if tenant is None or tenant.id != tenant_id:
-            raise PrivacyPolicyError("Tenant privacy policy not found")
-        tenant.settings = _set_settings_privacy_policy(tenant.settings, patch)
-        session.add(tenant)
-        return patch
-    if not scope_id:
-        raise PrivacyPolicyError(f"{clean_scope.capitalize()} privacy policy requires scope_id")
-    if clean_scope == "user":
-        current_settings = _user_settings(session, user_id=scope_id, tenant_id=tenant_id)
-        if current_settings is None:
-            raise PrivacyPolicyError("User privacy policy not found")
-        if _set_user_settings(session, user_id=scope_id, tenant_id=tenant_id, settings_payload=_set_settings_privacy_policy(current_settings, patch)) is None:
-            raise PrivacyPolicyError("User privacy policy not found")
-        return patch
-    if clean_scope == "group":
-        current_settings = _group_settings(session, group_id=scope_id, tenant_id=tenant_id)
-        if current_settings is None:
-            raise PrivacyPolicyError("Group privacy policy not found")
-        if _set_group_settings(session, group_id=scope_id, tenant_id=tenant_id, settings_payload=_set_settings_privacy_policy(current_settings, patch)) is None:
-            raise PrivacyPolicyError("Group privacy policy not found")
-        return patch
-    if clean_scope == "campaign":
-        provider = _campaign_policy_provider()
-        if provider is None:
-            raise PrivacyPolicyError("Campaign module is not installed")
-        campaign = _campaign_policy_context(session, tenant_id=tenant_id, campaign_id=scope_id)
-        settings_payload = _set_settings_privacy_policy(dict(campaign.settings or {}), patch)
-        try:
-            provider.set_campaign_settings(session, tenant_id=tenant_id, campaign_id=scope_id, settings=settings_payload)
-        except ValueError as exc:
-            raise PrivacyPolicyError("Campaign privacy policy not found") from exc
-        return patch
+        return _set_system_privacy_policy(session, policy)
+    patch = _privacy_policy_patch_payload(policy)
+    _validate_privacy_policy_patch_for_scope(
+        session,
+        tenant_id=tenant_id,
+        scope_type=clean_scope,
+        scope_id=scope_id,
+        patch=patch,
+    )
+    return _set_scoped_privacy_policy(
+        session,
+        tenant_id=tenant_id,
+        scope_type=clean_scope,
+        scope_id=scope_id,
+        patch=patch,
+    )
+
+
+def _set_system_privacy_policy(session: Session, policy: dict[str, Any] | None) -> dict[str, Any]:
+    item = get_system_settings(session)
+    validated = set_privacy_policy(item, PrivacyRetentionPolicy.model_validate(policy or {}))
+    session.add(item)
+    return validated.model_dump(mode="json")
+
+
+def _privacy_policy_patch_payload(policy: dict[str, Any] | None) -> dict[str, Any]:
+    return PrivacyRetentionPolicyPatch.model_validate(policy or {}).model_dump(mode="json", exclude_none=True)
+
+
+def _validate_privacy_policy_patch_for_scope(
+    session: Session,
+    *,
+    tenant_id: str,
+    scope_type: str,
+    scope_id: str | None,
+    patch: dict[str, Any],
+) -> None:
+    parent_scope_id = _parent_privacy_policy_scope_id(tenant_id=tenant_id, scope_type=scope_type, scope_id=scope_id)
+    parent = parent_privacy_policy(session, tenant_id=tenant_id, scope_type=scope_type, scope_id=parent_scope_id)
+    _validate_privacy_patch_against_parent(parent, patch)
+
+
+def _parent_privacy_policy_scope_id(*, tenant_id: str, scope_type: str, scope_id: str | None) -> str | None:
+    if scope_id:
+        return scope_id
+    if scope_type == "tenant":
+        return tenant_id
+    return None
+
+
+def _set_scoped_privacy_policy(
+    session: Session,
+    *,
+    tenant_id: str,
+    scope_type: str,
+    scope_id: str | None,
+    patch: dict[str, Any],
+) -> dict[str, Any]:
+    if scope_type == "tenant":
+        return _set_tenant_privacy_policy(session, tenant_id=tenant_id, scope_id=scope_id, patch=patch)
+    clean_scope_id = _required_privacy_policy_scope_id(scope_type, scope_id)
+    if scope_type == "user":
+        return _set_user_privacy_policy(session, tenant_id=tenant_id, user_id=clean_scope_id, patch=patch)
+    if scope_type == "group":
+        return _set_group_privacy_policy(session, tenant_id=tenant_id, group_id=clean_scope_id, patch=patch)
+    if scope_type == "campaign":
+        return _set_campaign_privacy_policy(session, tenant_id=tenant_id, campaign_id=clean_scope_id, patch=patch)
     raise PrivacyPolicyError("Privacy policy scope must be system, tenant, user, group or campaign")
 
 
+def _required_privacy_policy_scope_id(scope_type: str, scope_id: str | None) -> str:
+    if scope_id:
+        return scope_id
+    raise PrivacyPolicyError(f"{scope_type.capitalize()} privacy policy requires scope_id")
+
+
+def _set_tenant_privacy_policy(
+    session: Session,
+    *,
+    tenant_id: str,
+    scope_id: str | None,
+    patch: dict[str, Any],
+) -> dict[str, Any]:
+    tenant = session.get(Tenant, scope_id or tenant_id)
+    if tenant is None or tenant.id != tenant_id:
+        raise PrivacyPolicyError("Tenant privacy policy not found")
+    tenant.settings = _set_settings_privacy_policy(tenant.settings, patch)
+    session.add(tenant)
+    return patch
+
+
+def _set_user_privacy_policy(session: Session, *, tenant_id: str, user_id: str, patch: dict[str, Any]) -> dict[str, Any]:
+    current_settings = _user_settings(session, user_id=user_id, tenant_id=tenant_id)
+    if current_settings is None:
+        raise PrivacyPolicyError("User privacy policy not found")
+    settings_payload = _set_settings_privacy_policy(current_settings, patch)
+    if _set_user_settings(session, user_id=user_id, tenant_id=tenant_id, settings_payload=settings_payload) is None:
+        raise PrivacyPolicyError("User privacy policy not found")
+    return patch
+
+
+def _set_group_privacy_policy(session: Session, *, tenant_id: str, group_id: str, patch: dict[str, Any]) -> dict[str, Any]:
+    current_settings = _group_settings(session, group_id=group_id, tenant_id=tenant_id)
+    if current_settings is None:
+        raise PrivacyPolicyError("Group privacy policy not found")
+    settings_payload = _set_settings_privacy_policy(current_settings, patch)
+    if _set_group_settings(session, group_id=group_id, tenant_id=tenant_id, settings_payload=settings_payload) is None:
+        raise PrivacyPolicyError("Group privacy policy not found")
+    return patch
+
+
+def _set_campaign_privacy_policy(session: Session, *, tenant_id: str, campaign_id: str, patch: dict[str, Any]) -> dict[str, Any]:
+    provider = _campaign_policy_provider()
+    if provider is None:
+        raise PrivacyPolicyError("Campaign module is not installed")
+    campaign = _campaign_policy_context(session, tenant_id=tenant_id, campaign_id=campaign_id)
+    settings_payload = _set_settings_privacy_policy(dict(campaign.settings or {}), patch)
+    try:
+        provider.set_campaign_settings(session, tenant_id=tenant_id, campaign_id=campaign_id, settings=settings_payload)
+    except ValueError as exc:
+        raise PrivacyPolicyError("Campaign privacy policy not found") from exc
+    return patch
+
+
 def simulate_privacy_policy_change(
     session: Session,
     *,

src/govoplan_policy/backend/scheduling_privacy.py

@@ -0,0 +1,195 @@
+from __future__ import annotations
+
+from collections.abc import Mapping
+from dataclasses import dataclass
+from typing import Any, Literal, cast
+
+from sqlalchemy.orm import Session
+
+from govoplan_core.admin.models import SystemSettings
+from govoplan_core.admin.settings import SYSTEM_SETTINGS_ID
+from govoplan_core.core.policy import (
+    PolicySourceStep,
+    SchedulingParticipantPrivacyDecision,
+    SchedulingParticipantPrivacyRequest,
+    SchedulingParticipantVisibility,
+)
+from govoplan_core.tenancy.scope import Tenant
+
+
+SCHEDULING_PARTICIPANT_PRIVACY_SETTINGS_KEY = "scheduling_participant_privacy_policy"
+MAXIMUM_VISIBILITY_KEY = "maximum_visibility"
+
+_AGGREGATES_ONLY: SchedulingParticipantVisibility = "aggregates_only"
+_NAMES_AND_STATUSES: SchedulingParticipantVisibility = "names_and_statuses"
+
+
+@dataclass(frozen=True, slots=True)
+class _VisibilityCeiling:
+    value: SchedulingParticipantVisibility
+    configured: bool = False
+    issue: str | None = None
+
+
+def _visibility_ceiling(settings_payload: object) -> _VisibilityCeiling:
+    """Read one ceiling, treating malformed explicit policy as restrictive."""
+
+    if settings_payload is None:
+        return _VisibilityCeiling(value=_NAMES_AND_STATUSES)
+    if not isinstance(settings_payload, Mapping):
+        return _VisibilityCeiling(
+            value=_AGGREGATES_ONLY,
+            configured=True,
+            issue="invalid_settings_shape",
+        )
+    if SCHEDULING_PARTICIPANT_PRIVACY_SETTINGS_KEY not in settings_payload:
+        return _VisibilityCeiling(value=_NAMES_AND_STATUSES)
+
+    raw_policy = settings_payload.get(SCHEDULING_PARTICIPANT_PRIVACY_SETTINGS_KEY)
+    if not isinstance(raw_policy, Mapping) or MAXIMUM_VISIBILITY_KEY not in raw_policy:
+        return _VisibilityCeiling(
+            value=_AGGREGATES_ONLY,
+            configured=True,
+            issue="invalid_policy_shape",
+        )
+
+    raw_visibility = raw_policy.get(MAXIMUM_VISIBILITY_KEY)
+    if raw_visibility == _AGGREGATES_ONLY:
+        return _VisibilityCeiling(value=_AGGREGATES_ONLY, configured=True)
+    if raw_visibility == _NAMES_AND_STATUSES:
+        return _VisibilityCeiling(value=_NAMES_AND_STATUSES, configured=True)
+    return _VisibilityCeiling(
+        value=_AGGREGATES_ONLY,
+        configured=True,
+        issue="invalid_maximum_visibility",
+    )
+
+
+def _restrict_visibility(
+    current: SchedulingParticipantVisibility,
+    ceiling: SchedulingParticipantVisibility,
+) -> SchedulingParticipantVisibility:
+    if current == _AGGREGATES_ONLY or ceiling == _AGGREGATES_ONLY:
+        return _AGGREGATES_ONLY
+    return _NAMES_AND_STATUSES
+
+
+def _source_step(
+    *,
+    scope_type: Literal["system", "tenant"],
+    scope_id: str | None,
+    label: str,
+    ceiling: _VisibilityCeiling,
+) -> PolicySourceStep | None:
+    if not ceiling.configured:
+        return None
+    policy: dict[str, Any] = {MAXIMUM_VISIBILITY_KEY: ceiling.value}
+    if ceiling.issue is not None:
+        policy["configuration_status"] = "invalid_fail_closed"
+    return PolicySourceStep(
+        scope_type=scope_type,
+        scope_id=scope_id,
+        label=label,
+        applied_fields=(MAXIMUM_VISIBILITY_KEY,),
+        policy=policy,
+    )
+
+
+def _invalid_requested_visibility(value: object) -> bool:
+    return value not in {_AGGREGATES_ONLY, _NAMES_AND_STATUSES}
+
+
+class SqlSchedulingParticipantPrivacyPolicy:
+    """Resolve Scheduling roster disclosure against system and tenant ceilings.
+
+    Scheduling owns the request-level visibility setting. This provider is an
+    optional upper bound: absent policy preserves that setting, while an
+    explicit or malformed policy may only reduce it.
+    """
+
+    def resolve_scheduling_participant_visibility(
+        self,
+        session: object,
+        *,
+        request: SchedulingParticipantPrivacyRequest,
+    ) -> SchedulingParticipantPrivacyDecision:
+        db_session = cast(Session, session)
+        configuration_errors: list[dict[str, str]] = []
+
+        requested_invalid = _invalid_requested_visibility(request.requested_visibility)
+        requested_visibility: SchedulingParticipantVisibility = (
+            _AGGREGATES_ONLY if requested_invalid else request.requested_visibility
+        )
+        if requested_invalid:
+            configuration_errors.append({"scope": "request", "code": "invalid_requested_visibility"})
+
+        system_settings = db_session.get(SystemSettings, SYSTEM_SETTINGS_ID)
+        system_ceiling = _visibility_ceiling(system_settings.settings if system_settings is not None else None)
+        if system_ceiling.issue is not None:
+            configuration_errors.append({"scope": "system", "code": system_ceiling.issue})
+
+        tenant = db_session.get(Tenant, request.tenant_id)
+        if tenant is None:
+            tenant_ceiling = _VisibilityCeiling(
+                value=_AGGREGATES_ONLY,
+                configured=True,
+                issue="tenant_not_found",
+            )
+        else:
+            tenant_ceiling = _visibility_ceiling(tenant.settings)
+        if tenant_ceiling.issue is not None:
+            configuration_errors.append({"scope": "tenant", "code": tenant_ceiling.issue})
+
+        policy_ceiling = _restrict_visibility(system_ceiling.value, tenant_ceiling.value)
+        effective_visibility = _restrict_visibility(requested_visibility, policy_ceiling)
+
+        source_path = tuple(
+            step
+            for step in (
+                _source_step(
+                    scope_type="system",
+                    scope_id=None,
+                    label="System",
+                    ceiling=system_ceiling,
+                ),
+                _source_step(
+                    scope_type="tenant",
+                    scope_id=request.tenant_id,
+                    label="Tenant",
+                    ceiling=tenant_ceiling,
+                ),
+            )
+            if step is not None
+        )
+
+        reason: str | None = None
+        if configuration_errors:
+            reason = (
+                "Participant roster visibility was restricted because policy configuration "
+                "could not be validated."
+            )
+        elif effective_visibility != requested_visibility:
+            reason = "Participant roster visibility is restricted by policy."
+
+        return SchedulingParticipantPrivacyDecision(
+            effective_visibility=effective_visibility,
+            reason=reason,
+            source_path=source_path,
+            details={
+                "requested_visibility": request.requested_visibility,
+                "policy_ceiling": policy_ceiling,
+                "configured_scopes": [
+                    scope
+                    for scope, ceiling in (("system", system_ceiling), ("tenant", tenant_ceiling))
+                    if ceiling.configured
+                ],
+                "configuration_errors": configuration_errors,
+            },
+        )
+
+
+__all__ = [
+    "MAXIMUM_VISIBILITY_KEY",
+    "SCHEDULING_PARTICIPANT_PRIVACY_SETTINGS_KEY",
+    "SqlSchedulingParticipantPrivacyPolicy",
+]

tests/test_definition_governance.py

@@ -0,0 +1,147 @@
+from __future__ import annotations
+
+import unittest
+
+from govoplan_core.core.access import PrincipalRef
+from govoplan_core.core.policy import (
+    DefinitionGovernanceRequest,
+    DefinitionScopeRef,
+)
+from govoplan_policy.backend.definition_governance import (
+    DefinitionGovernancePolicyProvider,
+)
+
+
+class DefinitionGovernancePolicyTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.policy = DefinitionGovernancePolicyProvider()
+        self.actor = PrincipalRef(
+            account_id="account-1",
+            membership_id="user-1",
+            tenant_id="tenant-1",
+            group_ids=frozenset({"group-1"}),
+        )
+
+    def _resolve(self, **overrides):
+        values = {
+            "module_id": "dataflow",
+            "definition_ref": "pipeline:1",
+            "tenant_id": "tenant-1",
+            "definition_scope": DefinitionScopeRef(
+                "tenant",
+                "tenant-1",
+            ),
+            "target_scope": DefinitionScopeRef("tenant", "tenant-1"),
+            "definition_kind": "flow",
+            "action": "view",
+            "actor": self.actor,
+            "status": "active",
+            "inherit_to_lower_scopes": False,
+            "allow_run": True,
+            "allow_reuse": False,
+            "allow_automation": False,
+        }
+        values.update(overrides)
+        return self.policy.resolve_definition_action(
+            request=DefinitionGovernanceRequest(**values)
+        )
+
+    def test_system_definition_is_read_only_when_inherited(self) -> None:
+        view = self._resolve(
+            definition_scope=DefinitionScopeRef("system"),
+            inherit_to_lower_scopes=True,
+        )
+        edit = self._resolve(
+            definition_scope=DefinitionScopeRef("system"),
+            inherit_to_lower_scopes=True,
+            action="edit",
+        )
+
+        self.assertTrue(view.allowed)
+        self.assertFalse(edit.allowed)
+        self.assertEqual(
+            ["system", "tenant:tenant-1"],
+            [step.path for step in view.source_path],
+        )
+
+    def test_templates_cannot_run_or_be_automated(self) -> None:
+        run = self._resolve(
+            definition_kind="template",
+            action="run",
+            allow_run=True,
+        )
+        automated = self._resolve(
+            definition_kind="template",
+            action="automate",
+            allow_run=True,
+            allow_automation=True,
+        )
+
+        self.assertFalse(run.allowed)
+        self.assertFalse(automated.allowed)
+
+    def test_reuse_and_automation_are_independent_narrowing_grants(self) -> None:
+        reuse = self._resolve(action="reuse", allow_reuse=True)
+        automated = self._resolve(
+            action="automate",
+            allow_run=True,
+            allow_automation=False,
+        )
+
+        self.assertTrue(reuse.allowed)
+        self.assertFalse(automated.allowed)
+
+    def test_ancestor_limits_cannot_be_broadened(self) -> None:
+        decision = self._resolve(
+            action="reuse",
+            allow_reuse=True,
+            context={
+                "ancestor_limits": {"allow_reuse": False},
+                "ancestor_source": {
+                    "scope_type": "system",
+                    "label": "System template",
+                },
+            },
+        )
+
+        self.assertFalse(decision.allowed)
+        self.assertFalse(
+            decision.details["effective_limits"]["allow_reuse"]
+        )
+        self.assertEqual("system", decision.source_path[0].path)
+
+    def test_ancestor_inheritance_limit_controls_visibility(self) -> None:
+        decision = self._resolve(
+            definition_scope=DefinitionScopeRef("system"),
+            inherit_to_lower_scopes=True,
+            context={
+                "ancestor_limits": {
+                    "inherit_to_lower_scopes": False,
+                },
+                "ancestor_source": {
+                    "scope_type": "system",
+                    "label": "Restricted ancestor",
+                },
+            },
+        )
+
+        self.assertFalse(decision.allowed)
+        self.assertEqual(
+            "The system definition is not inherited by lower scopes.",
+            decision.reason,
+        )
+
+    def test_group_and_user_scopes_are_visible_only_to_the_actor(self) -> None:
+        group = self._resolve(
+            definition_scope=DefinitionScopeRef("group", "group-1")
+        )
+        user = self._resolve(
+            definition_scope=DefinitionScopeRef("user", "user-2")
+        )
+
+        self.assertTrue(group.allowed)
+        self.assertFalse(user.allowed)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_policy_hierarchy.py

@@ -2,6 +2,14 @@ from __future__ import annotations
 
 import unittest
 
+from govoplan_policy.backend.retention import (
+    PrivacyRetentionPolicy,
+    PrivacyRetentionPolicyPatch,
+    PrivacyPolicyError,
+    _parent_privacy_policy_scope_id,
+    _privacy_policy_patch_payload,
+    _required_privacy_policy_scope_id,
+)
 from govoplan_policy.backend.hierarchy import (
     PolicyRestrictionRule,
     simulate_hierarchical_policy_change,
@@ -10,6 +18,36 @@ from govoplan_policy.backend.hierarchy import (
 
 
 class PolicyHierarchyTests(unittest.TestCase):
+    def test_privacy_policy_parent_scope_ids_follow_scope_precedence(self) -> None:
+        self.assertEqual(
+            "tenant-1",
+            _parent_privacy_policy_scope_id(tenant_id="tenant-1", scope_type="tenant", scope_id=None),
+        )
+        self.assertIsNone(_parent_privacy_policy_scope_id(tenant_id="tenant-1", scope_type="user", scope_id=None))
+        self.assertIsNone(_parent_privacy_policy_scope_id(tenant_id="tenant-1", scope_type="group", scope_id=None))
+        self.assertEqual(
+            "campaign-1",
+            _parent_privacy_policy_scope_id(tenant_id="tenant-1", scope_type="campaign", scope_id="campaign-1"),
+        )
+
+    def test_privacy_policy_patch_payload_validates_and_removes_unset_fields(self) -> None:
+        payload = _privacy_policy_patch_payload({"audit_detail_level": "minimal", "generated_eml_retention_days": None})
+
+        self.assertEqual({"audit_detail_level": "minimal"}, payload)
+
+    def test_privacy_policy_models_extend_shared_schema_contract(self) -> None:
+        policy = PrivacyRetentionPolicy.model_validate({"generated_eml_retention_days": ""})
+        patch = PrivacyRetentionPolicyPatch.model_validate({"allow_lower_level_limits": {"audit_detail_level": False}})
+
+        self.assertIsNone(policy.generated_eml_retention_days)
+        self.assertEqual({"audit_detail_level": False}, patch.allow_lower_level_limits)
+
+    def test_required_privacy_policy_scope_id_reports_missing_scope(self) -> None:
+        with self.assertRaises(PrivacyPolicyError) as captured:
+            _required_privacy_policy_scope_id("group", None)
+
+        self.assertEqual("Group privacy policy requires scope_id", str(captured.exception))
+
     def test_parent_locks_block_lower_level_field_changes_and_limit_reenable(self) -> None:
         issues = validate_hierarchical_policy_patch(
             parent_policy={"retention_days": 30},

tests/test_policy_module_contract.py

@@ -4,6 +4,13 @@ import pathlib
 import tomllib
 import unittest
 
+from govoplan_core.core.policy import (
+    CAPABILITY_POLICY_DEFINITION_GOVERNANCE,
+    CAPABILITY_POLICY_PRIVACY_RETENTION,
+    CAPABILITY_POLICY_SCHEDULING_PARTICIPANT_PRIVACY,
+)
+from govoplan_policy.backend.manifest import manifest
+
 
 ROOT = pathlib.Path(__file__).resolve().parents[1]
 
@@ -13,7 +20,7 @@ class PolicyModuleContractTests(unittest.TestCase):
         project = tomllib.loads((ROOT / "pyproject.toml").read_text(encoding="utf-8"))["project"]
         dependencies = tuple(project["dependencies"])
 
-        self.assertIn("govoplan-core>=0.1.6", dependencies)
+        self.assertTrue(any(item.startswith("govoplan-core>=") for item in dependencies))
         self.assertFalse(any(item.startswith("govoplan-access") for item in dependencies))
 
     def test_policy_source_does_not_import_access_implementation(self) -> None:
@@ -25,6 +32,16 @@ class PolicyModuleContractTests(unittest.TestCase):
 
         self.assertEqual([], offenders)
 
+    def test_policy_manifest_exposes_policy_capabilities(self) -> None:
+        self.assertEqual(
+            {
+                CAPABILITY_POLICY_DEFINITION_GOVERNANCE,
+                CAPABILITY_POLICY_PRIVACY_RETENTION,
+                CAPABILITY_POLICY_SCHEDULING_PARTICIPANT_PRIVACY,
+            },
+            set(manifest.capability_factories),
+        )
+
 
 if __name__ == "__main__":
     unittest.main()

tests/test_scheduling_privacy.py

@@ -0,0 +1,178 @@
+from __future__ import annotations
+
+import unittest
+
+from sqlalchemy import create_engine
+from sqlalchemy.orm import Session
+
+from govoplan_core.admin.models import SystemSettings
+from govoplan_core.core.modules import ModuleContext
+from govoplan_core.core.policy import (
+    CAPABILITY_POLICY_SCHEDULING_PARTICIPANT_PRIVACY,
+    SchedulingParticipantPrivacyPolicy,
+    SchedulingParticipantPrivacyRequest,
+)
+from govoplan_core.db.base import Base
+from govoplan_core.tenancy.scope import Tenant, scope_registry
+from govoplan_policy.backend.manifest import manifest
+from govoplan_policy.backend.scheduling_privacy import (
+    SCHEDULING_PARTICIPANT_PRIVACY_SETTINGS_KEY,
+    SqlSchedulingParticipantPrivacyPolicy,
+)
+
+
+def _policy_settings(maximum_visibility: object) -> dict[str, object]:
+    return {
+        SCHEDULING_PARTICIPANT_PRIVACY_SETTINGS_KEY: {
+            "maximum_visibility": maximum_visibility,
+        }
+    }
+
+
+class SchedulingParticipantPrivacyTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.engine = create_engine("sqlite+pysqlite:///:memory:")
+        Base.metadata.create_all(bind=self.engine, tables=[SystemSettings.__table__])
+        scope_registry.metadata.create_all(bind=self.engine, tables=[Tenant.__table__])
+        self.session = Session(self.engine)
+        self.provider = SqlSchedulingParticipantPrivacyPolicy()
+
+    def tearDown(self) -> None:
+        self.session.close()
+        self.engine.dispose()
+
+    def _add_settings(
+        self,
+        *,
+        system: dict[str, object] | None = None,
+        tenant: dict[str, object] | None = None,
+    ) -> None:
+        self.session.add(SystemSettings(id="global", settings=system or {}))
+        self.session.add(Tenant(id="tenant-1", slug="tenant-1", name="Tenant 1", settings=tenant or {}))
+        self.session.flush()
+
+    def _resolve(self, requested_visibility: str = "names_and_statuses"):
+        return self.provider.resolve_scheduling_participant_visibility(
+            self.session,
+            request=SchedulingParticipantPrivacyRequest(
+                tenant_id="tenant-1",
+                scheduling_request_id="request-1",
+                participant_id="participant-1",
+                requested_visibility=requested_visibility,  # type: ignore[arg-type]
+                actor_user_id="user-1",
+            ),
+        )
+
+    def test_manifest_registers_the_core_capability(self) -> None:
+        factory = manifest.capability_factories[CAPABILITY_POLICY_SCHEDULING_PARTICIPANT_PRIVACY]
+
+        provider = factory(ModuleContext(registry=object(), settings=object()))
+
+        self.assertIsInstance(provider, SchedulingParticipantPrivacyPolicy)
+
+    def test_missing_policy_preserves_scheduling_visibility(self) -> None:
+        self._add_settings()
+
+        decision = self._resolve()
+
+        self.assertEqual("names_and_statuses", decision.effective_visibility)
+        self.assertIsNone(decision.reason)
+        self.assertEqual((), decision.source_path)
+        self.assertEqual([], decision.details["configured_scopes"])
+
+    def test_missing_system_settings_is_an_unrestricted_read_only_default(self) -> None:
+        self.session.add(Tenant(id="tenant-1", slug="tenant-1", name="Tenant 1", settings={}))
+        self.session.flush()
+
+        decision = self._resolve()
+
+        self.assertEqual("names_and_statuses", decision.effective_visibility)
+        self.assertIsNone(self.session.get(SystemSettings, "global"))
+
+    def test_system_ceiling_restricts_participant_roster(self) -> None:
+        self._add_settings(system=_policy_settings("aggregates_only"))
+
+        decision = self._resolve()
+
+        self.assertEqual("aggregates_only", decision.effective_visibility)
+        self.assertEqual("Participant roster visibility is restricted by policy.", decision.reason)
+        self.assertEqual(["system"], [step.path for step in decision.source_path])
+        self.assertEqual("aggregates_only", decision.source_path[0].policy["maximum_visibility"])
+
+    def test_tenant_ceiling_can_narrow_system_but_cannot_widen_it(self) -> None:
+        self._add_settings(
+            system=_policy_settings("aggregates_only"),
+            tenant=_policy_settings("names_and_statuses"),
+        )
+
+        decision = self._resolve()
+
+        self.assertEqual("aggregates_only", decision.effective_visibility)
+        self.assertEqual(["system", "tenant:tenant-1"], [step.path for step in decision.source_path])
+        self.assertEqual("aggregates_only", decision.details["policy_ceiling"])
+
+    def test_policy_never_broadens_an_aggregate_only_request(self) -> None:
+        self._add_settings(
+            system=_policy_settings("names_and_statuses"),
+            tenant=_policy_settings("names_and_statuses"),
+        )
+
+        decision = self._resolve("aggregates_only")
+
+        self.assertEqual("aggregates_only", decision.effective_visibility)
+        self.assertIsNone(decision.reason)
+
+    def test_invalid_explicit_policy_fails_closed_without_echoing_value(self) -> None:
+        self._add_settings(tenant=_policy_settings("surprise"))
+
+        decision = self._resolve()
+
+        self.assertEqual("aggregates_only", decision.effective_visibility)
+        self.assertIn("could not be validated", decision.reason or "")
+        self.assertEqual(
+            [{"scope": "tenant", "code": "invalid_maximum_visibility"}],
+            decision.details["configuration_errors"],
+        )
+        self.assertNotIn("surprise", repr(decision.to_dict()))
+        self.assertEqual("invalid_fail_closed", decision.source_path[0].policy["configuration_status"])
+
+    def test_invalid_policy_shape_fails_closed(self) -> None:
+        self._add_settings(
+            system={SCHEDULING_PARTICIPANT_PRIVACY_SETTINGS_KEY: "not-an-object"},
+        )
+
+        decision = self._resolve()
+
+        self.assertEqual("aggregates_only", decision.effective_visibility)
+        self.assertEqual(
+            [{"scope": "system", "code": "invalid_policy_shape"}],
+            decision.details["configuration_errors"],
+        )
+
+    def test_missing_tenant_fails_closed(self) -> None:
+        self.session.add(SystemSettings(id="global", settings={}))
+        self.session.flush()
+
+        decision = self._resolve()
+
+        self.assertEqual("aggregates_only", decision.effective_visibility)
+        self.assertEqual(
+            [{"scope": "tenant", "code": "tenant_not_found"}],
+            decision.details["configuration_errors"],
+        )
+        self.assertEqual(["tenant:tenant-1"], [step.path for step in decision.source_path])
+
+    def test_invalid_requested_visibility_fails_closed(self) -> None:
+        self._add_settings()
+
+        decision = self._resolve("invalid")
+
+        self.assertEqual("aggregates_only", decision.effective_visibility)
+        self.assertEqual(
+            [{"scope": "request", "code": "invalid_requested_visibility"}],
+            decision.details["configuration_errors"],
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()

webui/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@govoplan/policy-webui",
-  "version": "0.1.7",
+  "version": "0.1.9",
   "private": true,
   "type": "module",
   "main": "src/index.ts",
@@ -13,7 +13,7 @@
     }
   },
   "peerDependencies": {
-    "@govoplan/core-webui": "^0.1.7",
+    "@govoplan/core-webui": "^0.1.9",
     "lucide-react": "^1.23.0",
     "react": "^19.0.0",
     "react-dom": "^19.0.0",

webui/src/module.ts

@@ -7,6 +7,7 @@ const policyAdminSections: AdminSectionsUiCapability = {
   sections: [
     {
       id: "system-retention",
+      surfaceId: "policy.admin.system-retention",
       label: "Retention",
       group: "SYSTEM",
       order: 80,
@@ -19,6 +20,7 @@ const policyAdminSections: AdminSectionsUiCapability = {
     },
     {
       id: "tenant-retention",
+      surfaceId: "policy.admin.tenant-retention",
       label: "Retention",
       group: "TENANT",
       order: 80,
@@ -31,6 +33,7 @@ const policyAdminSections: AdminSectionsUiCapability = {
     },
     {
       id: "tenant-group-retention",
+      surfaceId: "policy.admin.group-retention",
       label: "Retention",
       group: "GROUP",
       order: 30,
@@ -43,6 +46,7 @@ const policyAdminSections: AdminSectionsUiCapability = {
     },
     {
       id: "tenant-user-retention",
+      surfaceId: "policy.admin.user-retention",
       label: "Retention",
       group: "USER",
       order: 30,
@@ -59,8 +63,14 @@ const policyAdminSections: AdminSectionsUiCapability = {
 export const policyModule: PlatformWebModule = {
   id: "policy",
   label: "Policy",
-  version: "0.1.6",
+  version: "0.1.9",
   dependencies: ["access", "admin"],
+  viewSurfaces: [
+    { id: "policy.admin.system-retention", moduleId: "policy", kind: "section", label: "System retention", order: 80 },
+    { id: "policy.admin.tenant-retention", moduleId: "policy", kind: "section", label: "Tenant retention", order: 80 },
+    { id: "policy.admin.group-retention", moduleId: "policy", kind: "section", label: "Group retention", order: 80 },
+    { id: "policy.admin.user-retention", moduleId: "policy", kind: "section", label: "User retention", order: 80 }
+  ],
   uiCapabilities: {
     "admin.sections": policyAdminSections
   }