Declare policy View surfaces

README.md

@@ -1,11 +1,38 @@
 # GovOPlaN Policy
 
-`govoplan-policy` owns policy and retention API route contributions during the
-GovOPlaN module split.
+<!-- govoplan-repository-type:start -->
+**Repository type:** module (platform).
+<!-- govoplan-repository-type:end -->
 
-The current package delegates to the legacy access administration
-implementation while route ownership is separated before model migration.
+`govoplan-policy` owns policy and retention API route contributions and the
+retention administration WebUI sections during the GovOPlaN module split.
+
+The `@govoplan/policy-webui` package contributes system, tenant, group, and
+user retention sections through the shared `admin.sections` UI capability. The
+admin shell does not render retention policy panels unless this module is
+installed and enabled.
 
 Policy decision and provenance payloads use the shared kernel DTOs documented
 in [docs/POLICY_DECISION_PROVENANCE.md](docs/POLICY_DECISION_PROVENANCE.md)
 and `/mnt/DATA/git/govoplan-core/docs/POLICY_CONTRACTS.md`.
+
+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

@@ -9,10 +9,45 @@ Privacy retention implementation lives in this module at
 dispatch to the active policy module without owning policy logic in core. Core
 does not import this implementation as a hidden fallback when policy is
 disabled.
+
+Reusable hierarchical policy validation lives in
+`govoplan_policy.backend.hierarchy`. Policy families should use that helper for
+parent locks, lower-level override ceilings, "more restrictive only" checks,
+and read-only simulations before destructive or limiting changes are saved.
+Domain modules keep their own policy fields and restriction rules, but the
+decision shape and simulation payload stay consistent.
 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:
@@ -47,6 +82,18 @@ GET /api/v1/admin/privacy-retention/policies/{scope_type}/explain
 
 The response contains `decision`, `effective_policy`, `parent_policy`,
 `effective_policy_sources`, `parent_policy_sources`, and `blocked_fields`.
+
+Retention policy also exposes a write-preflight endpoint:
+
+```text
+POST /api/v1/admin/privacy-retention/policies/{scope}/simulate
+```
+
+The request body is the same as the write endpoint. The response contains a
+`simulation` object with `allowed`, `changed_fields`, `issues`,
+`before_policy`, `requested_policy`, and a shared `PolicyDecision` payload.
+The endpoint never writes policy state and is intended for UI validation before
+operators attempt destructive or limiting changes.
 Clients can use `blocked_fields` to disable controls before a save attempt.
 
 ## UI Expectations

package.json

@@ -0,0 +1,32 @@
+{
+  "name": "@govoplan/policy-webui",
+  "version": "0.1.9",
+  "private": true,
+  "type": "module",
+  "main": "webui/src/index.ts",
+  "module": "webui/src/index.ts",
+  "types": "webui/src/index.ts",
+  "exports": {
+    ".": {
+      "types": "./webui/src/index.ts",
+      "import": "./webui/src/index.ts"
+    }
+  },
+  "files": [
+    "webui/src",
+    "README.md",
+    "LICENSE"
+  ],
+  "peerDependencies": {
+    "@govoplan/core-webui": "^0.1.9",
+    "lucide-react": "^1.23.0",
+    "react": "^19.0.0",
+    "react-dom": "^19.0.0",
+    "react-router-dom": "^7.1.1"
+  },
+  "peerDependenciesMeta": {
+    "@govoplan/core-webui": {
+      "optional": true
+    }
+  }
+}

pyproject.toml

@@ -4,14 +4,13 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "govoplan-policy"
-version = "0.1.6"
+version = "0.1.9"
 description = "GovOPlaN policy platform module."
 readme = "README.md"
 requires-python = ">=3.12"
 authors = [{ name = "GovOPlaN" }]
 dependencies = [
-  "govoplan-core>=0.1.6",
-  "govoplan-access>=0.1.6",
+  "govoplan-core>=0.1.9",
 ]
 
 [tool.setuptools.packages.find]
@@ -22,4 +21,3 @@ govoplan_policy = ["py.typed"]
 
 [project.entry-points."govoplan.modules"]
 policy = "govoplan_policy.backend.manifest:get_manifest"
-

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,
@@ -24,6 +24,7 @@ from govoplan_policy.backend.retention import (
     parent_privacy_policy,
     parent_privacy_policy_sources,
     set_privacy_policy_for_scope,
+    simulate_privacy_policy_change,
 )
 
 from .schemas import (
@@ -31,6 +32,7 @@ from .schemas import (
     PrivacyRetentionPolicyItem,
     PrivacyRetentionPolicyScopeRequest,
     PrivacyRetentionPolicyScopeResponse,
+    PrivacyRetentionPolicySimulationResponse,
     RetentionRunRequest,
     RetentionRunResponse,
 )
@@ -200,6 +202,32 @@ def explain_privacy_retention_policy(
         raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc
 
 
+@router.post("/privacy-retention/policies/{scope_type}/simulate", response_model=PrivacyRetentionPolicySimulationResponse)
+def simulate_privacy_retention_policy(
+    scope_type: str,
+    payload: PrivacyRetentionPolicyScopeRequest,
+    scope_id: str | None = Query(default=None),
+    session: Session = Depends(get_session),
+    principal: ApiPrincipal = Depends(get_api_principal),
+):
+    clean_scope = scope_type.strip().casefold()
+    _require_privacy_policy_write(principal, clean_scope)
+    try:
+        return PrivacyRetentionPolicySimulationResponse(
+            scope_type=clean_scope,
+            scope_id=scope_id,
+            simulation=simulate_privacy_policy_change(
+                session,
+                tenant_id=principal.tenant_id,
+                scope_type=clean_scope,
+                scope_id=scope_id,
+                policy=payload.policy.model_dump(mode="json", exclude_none=True),
+            ),
+        )
+    except PrivacyPolicyError as exc:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc
+
+
 def _blocked_privacy_retention_fields(parent) -> list[str]:
     if parent is None:
         return []

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")
 
@@ -121,6 +52,12 @@ class PrivacyRetentionPolicyExplainResponse(BaseModel):
     blocked_fields: list[str] = Field(default_factory=list)
 
 
+class PrivacyRetentionPolicySimulationResponse(BaseModel):
+    scope_type: Literal["system", "tenant", "user", "group", "campaign"]
+    scope_id: str | None = None
+    simulation: dict[str, Any]
+
+
 class RetentionRunRequest(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

@@ -0,0 +1,153 @@
+from __future__ import annotations
+
+from collections.abc import Callable, Mapping, Sequence
+from dataclasses import dataclass, field
+from typing import Any, Literal
+
+from govoplan_core.core.policy import PolicyDecision, PolicySourceStep
+
+PolicyIssueSeverity = Literal["blocker", "warning", "info"]
+RestrictionCheck = Callable[[Any, Any], bool]
+
+
+@dataclass(frozen=True, slots=True)
+class PolicyValidationIssue:
+    code: str
+    message: str
+    field: str | None = None
+    severity: PolicyIssueSeverity = "blocker"
+
+    def to_dict(self) -> dict[str, object]:
+        payload: dict[str, object] = {
+            "severity": self.severity,
+            "code": self.code,
+            "message": self.message,
+        }
+        if self.field is not None:
+            payload["field"] = self.field
+        return payload
+
+
+@dataclass(frozen=True, slots=True)
+class PolicyRestrictionRule:
+    field: str
+    is_more_restrictive_or_equal: RestrictionCheck
+    less_restrictive_message: str
+
+
+@dataclass(frozen=True, slots=True)
+class PolicySimulationResult:
+    allowed: bool
+    changed_fields: tuple[str, ...]
+    issues: tuple[PolicyValidationIssue, ...] = ()
+    before_policy: Mapping[str, Any] = field(default_factory=dict)
+    requested_policy: Mapping[str, Any] = field(default_factory=dict)
+    decision: PolicyDecision | None = None
+
+    def to_dict(self) -> dict[str, object]:
+        return {
+            "allowed": self.allowed,
+            "changed_fields": list(self.changed_fields),
+            "issues": [issue.to_dict() for issue in self.issues],
+            "before_policy": dict(self.before_policy),
+            "requested_policy": dict(self.requested_policy),
+            "decision": self.decision.to_dict() if self.decision is not None else None,
+        }
+
+
+def validate_hierarchical_policy_patch(
+    *,
+    parent_policy: Mapping[str, Any],
+    patch: Mapping[str, Any],
+    field_keys: Sequence[str],
+    parent_allow_lower_level_limits: Mapping[str, bool],
+    restriction_rules: Sequence[PolicyRestrictionRule] = (),
+    locked_field_message: Callable[[str], str] | None = None,
+    relock_message: Callable[[str], str] | None = None,
+) -> tuple[PolicyValidationIssue, ...]:
+    issues: list[PolicyValidationIssue] = []
+    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_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_name,
+                message=lock_message(field_name),
+            ))
+
+    patch_allow = patch.get("allow_lower_level_limits")
+    if isinstance(patch_allow, Mapping):
+        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",
+                    field=clean_field,
+                    message=reenable_message(clean_field),
+                ))
+
+    for rule in restriction_rules:
+        if rule.field not in patch or patch.get(rule.field) is None:
+            continue
+        parent_value = parent_policy.get(rule.field)
+        requested_value = patch.get(rule.field)
+        if not rule.is_more_restrictive_or_equal(parent_value, requested_value):
+            issues.append(PolicyValidationIssue(
+                code="less_restrictive_than_parent",
+                field=rule.field,
+                message=rule.less_restrictive_message,
+            ))
+
+    return tuple(issues)
+
+
+def simulate_hierarchical_policy_change(
+    *,
+    parent_policy: Mapping[str, Any],
+    current_policy: Mapping[str, Any],
+    patch: Mapping[str, Any],
+    field_keys: Sequence[str],
+    parent_allow_lower_level_limits: Mapping[str, bool],
+    restriction_rules: Sequence[PolicyRestrictionRule] = (),
+    source_path: Sequence[PolicySourceStep] = (),
+    locked_field_message: Callable[[str], str] | None = None,
+    relock_message: Callable[[str], str] | None = None,
+) -> PolicySimulationResult:
+    issues = validate_hierarchical_policy_patch(
+        parent_policy=parent_policy,
+        patch=patch,
+        field_keys=field_keys,
+        parent_allow_lower_level_limits=parent_allow_lower_level_limits,
+        restriction_rules=restriction_rules,
+        locked_field_message=locked_field_message,
+        relock_message=relock_message,
+    )
+    changed_fields = tuple(
+        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(
+        allowed=allowed,
+        reason=None if allowed else "Requested policy change is blocked by parent policy constraints.",
+        source_path=tuple(source_path),
+        requirements=tuple(issue.field or issue.code for issue in issues if issue.severity == "blocker"),
+        details={"issues": [issue.to_dict() for issue in issues], "changed_fields": list(changed_fields)},
+    )
+    return PolicySimulationResult(
+        allowed=allowed,
+        changed_fields=changed_fields,
+        issues=issues,
+        before_policy=dict(current_policy),
+        requested_policy=dict(patch),
+        decision=decision,
+    )

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 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,14 +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.6",
+    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
@@ -23,45 +23,28 @@ from govoplan_core.core.access import (
     AuditRecordRef,
     AuditRetentionProvider,
 )
-from govoplan_core.core.policy import policy_source_step
+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 (
+    PolicyRestrictionRule,
+    simulate_hierarchical_policy_change,
+    validate_hierarchical_policy_patch,
+)
 
 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",
@@ -86,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",
@@ -115,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:
@@ -140,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):
@@ -277,26 +238,46 @@ def _merge_privacy_policy(parent: PrivacyRetentionPolicy, patch: dict[str, Any])
     return PrivacyRetentionPolicy.model_validate(payload)
 
 
+def _parent_allow_lower_level_limits(parent_payload: dict[str, Any]) -> dict[str, bool]:
+    return {**default_allow_lower_level_limits(), **(parent_payload.get("allow_lower_level_limits") or {})}
+
+
+def _privacy_restriction_rules() -> tuple[PolicyRestrictionRule, ...]:
+    return (
+        PolicyRestrictionRule(
+            field="store_raw_campaign_json",
+            is_more_restrictive_or_equal=lambda parent_value, requested_value: not (requested_value is True and parent_value is False),
+            less_restrictive_message="Raw campaign JSON storage cannot be re-enabled below a parent policy that disables it.",
+        ),
+        *(
+            PolicyRestrictionRule(
+                field=key,
+                is_more_restrictive_or_equal=lambda parent_value, requested_value: parent_value is None or int(requested_value) <= int(parent_value),
+                less_restrictive_message=f"{key} cannot be less restrictive than the parent retention policy.",
+            )
+            for key in RETENTION_DAY_KEYS
+        ),
+        PolicyRestrictionRule(
+            field="audit_detail_level",
+            is_more_restrictive_or_equal=lambda parent_value, requested_value: AUDIT_DETAIL_LEVEL_ORDER[str(requested_value)] >= AUDIT_DETAIL_LEVEL_ORDER[str(parent_value or "full")],
+            less_restrictive_message="Audit detail level cannot be less restrictive than the parent retention policy.",
+        ),
+    )
+
+
 def _validate_privacy_patch_against_parent(parent: PrivacyRetentionPolicy, patch: dict[str, Any]) -> None:
     parent_payload = parent.model_dump(mode="json")
-    parent_allow = {**default_allow_lower_level_limits(), **(parent_payload.get("allow_lower_level_limits") or {})}
-    for key in RETENTION_POLICY_FIELD_KEYS:
-        if key in patch and patch.get(key) is not None and not parent_allow[key]:
-            raise PrivacyPolicyError(f"{key} is locked by the parent retention policy.")
-    patch_allow = patch.get("allow_lower_level_limits") or {}
-    for key, allowed in patch_allow.items():
-        if allowed and not parent_allow[key]:
-            raise PrivacyPolicyError(f"{key} limiting cannot be re-enabled below a parent retention policy lock.")
-    if patch.get("store_raw_campaign_json") is True and not parent.store_raw_campaign_json:
-        raise PrivacyPolicyError("Raw campaign JSON storage cannot be re-enabled below a parent policy that disables it.")
-    for key in RETENTION_DAY_KEYS:
-        value = patch.get(key)
-        parent_value = parent_payload.get(key)
-        if value is not None and parent_value is not None and int(value) > int(parent_value):
-            raise PrivacyPolicyError(f"{key} cannot be less restrictive than the parent retention policy.")
-    detail_level = patch.get("audit_detail_level")
-    if detail_level and AUDIT_DETAIL_LEVEL_ORDER[detail_level] < AUDIT_DETAIL_LEVEL_ORDER[parent.audit_detail_level]:
-        raise PrivacyPolicyError("Audit detail level cannot be less restrictive than the parent retention policy.")
+    issues = validate_hierarchical_policy_patch(
+        parent_policy=parent_payload,
+        patch=patch,
+        field_keys=RETENTION_POLICY_FIELD_KEYS,
+        parent_allow_lower_level_limits=_parent_allow_lower_level_limits(parent_payload),
+        restriction_rules=_privacy_restriction_rules(),
+        locked_field_message=lambda key: f"{key} is locked by the parent retention policy.",
+        relock_message=lambda key: f"{key} limiting cannot be re-enabled below a parent retention policy lock.",
+    )
+    if issues:
+        raise PrivacyPolicyError(issues[0].message)
 
 
 def _set_settings_privacy_policy(settings_payload: dict[str, Any] | None, policy: dict[str, Any]) -> dict[str, Any]:
@@ -499,49 +480,175 @@ 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,
+    *,
+    tenant_id: str,
+    scope_type: str,
+    scope_id: str | None = None,
+    policy: dict[str, Any] | None = None,
+) -> dict[str, Any]:
+    clean_scope = scope_type.strip().casefold()
+    patch = PrivacyRetentionPolicyPatch.model_validate(policy or {}).model_dump(mode="json", exclude_none=True)
+    current = get_privacy_policy_for_scope(session, tenant_id=tenant_id, scope_type=clean_scope, scope_id=scope_id)
+    if clean_scope == "system":
+        parent_payload = PrivacyRetentionPolicy.model_validate(current).model_dump(mode="json")
+        parent_allow = default_allow_lower_level_limits()
+        parent_sources = effective_privacy_policy_sources(session)
+    else:
+        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),
+        )
+        parent_payload = parent.model_dump(mode="json")
+        parent_allow = _parent_allow_lower_level_limits(parent_payload)
+        parent_sources = parent_privacy_policy_sources(
+            session,
+            tenant_id=tenant_id,
+            scope_type=clean_scope,
+            scope_id=scope_id or (tenant_id if clean_scope == "tenant" else None),
+        )
+
+    simulation = simulate_hierarchical_policy_change(
+        parent_policy=parent_payload,
+        current_policy=current,
+        patch=patch,
+        field_keys=RETENTION_POLICY_FIELD_KEYS,
+        parent_allow_lower_level_limits=parent_allow,
+        restriction_rules=() if clean_scope == "system" else _privacy_restriction_rules(),
+        source_path=[PolicySourceStep.from_mapping(source) for source in parent_sources],
+        locked_field_message=lambda key: f"{key} is locked by the parent retention policy.",
+        relock_message=lambda key: f"{key} limiting cannot be re-enabled below a parent retention policy lock.",
+    )
+    return simulation.to_dict()
+
+
 def sanitize_audit_details_for_policy(session: Session, details: dict[str, Any]) -> dict[str, Any]:
     policy = privacy_policy_from_session(session)
     if policy.audit_detail_level == "full":
@@ -723,6 +830,9 @@ class SqlPrivacyRetentionService:
     def set_privacy_policy_for_scope(self, *args: Any, **kwargs: Any) -> Any:
         return set_privacy_policy_for_scope(*args, **kwargs)
 
+    def simulate_privacy_policy_change(self, *args: Any, **kwargs: Any) -> Any:
+        return simulate_privacy_policy_change(*args, **kwargs)
+
     def sanitize_audit_details_for_policy(self, *args: Any, **kwargs: Any) -> Any:
         return sanitize_audit_details_for_policy(*args, **kwargs)
 

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

@@ -0,0 +1,105 @@
+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,
+    validate_hierarchical_policy_patch,
+)
+
+
+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},
+            patch={"retention_days": 20, "allow_lower_level_limits": {"retention_days": True}},
+            field_keys=("retention_days",),
+            parent_allow_lower_level_limits={"retention_days": False},
+        )
+
+        self.assertEqual(["field_locked_by_parent", "lower_level_limit_locked_by_parent"], [issue.code for issue in issues])
+        self.assertEqual(["retention_days", "retention_days"], [issue.field for issue in issues])
+
+    def test_restriction_rules_block_less_restrictive_patch(self) -> None:
+        issues = validate_hierarchical_policy_patch(
+            parent_policy={"retention_days": 30},
+            patch={"retention_days": 45},
+            field_keys=("retention_days",),
+            parent_allow_lower_level_limits={"retention_days": True},
+            restriction_rules=(
+                PolicyRestrictionRule(
+                    field="retention_days",
+                    is_more_restrictive_or_equal=lambda parent, requested: int(requested) <= int(parent),
+                    less_restrictive_message="retention_days cannot be widened",
+                ),
+            ),
+        )
+
+        self.assertEqual(1, len(issues))
+        self.assertEqual("less_restrictive_than_parent", issues[0].code)
+        self.assertEqual("retention_days cannot be widened", issues[0].message)
+
+    def test_policy_simulation_returns_decision_payload(self) -> None:
+        simulation = simulate_hierarchical_policy_change(
+            parent_policy={"retention_days": 30},
+            current_policy={"retention_days": 30},
+            patch={"retention_days": 45},
+            field_keys=("retention_days",),
+            parent_allow_lower_level_limits={"retention_days": True},
+            restriction_rules=(
+                PolicyRestrictionRule(
+                    field="retention_days",
+                    is_more_restrictive_or_equal=lambda parent, requested: int(requested) <= int(parent),
+                    less_restrictive_message="retention_days cannot be widened",
+                ),
+            ),
+        )
+
+        self.assertFalse(simulation.allowed)
+        self.assertEqual(("retention_days",), simulation.changed_fields)
+        self.assertIsNotNone(simulation.decision)
+        self.assertEqual(("retention_days",), simulation.decision.requirements)
+        self.assertEqual("Requested policy change is blocked by parent policy constraints.", simulation.decision.reason)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_policy_module_contract.py

@@ -0,0 +1,47 @@
+from __future__ import annotations
+
+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]
+
+
+class PolicyModuleContractTests(unittest.TestCase):
+    def test_policy_package_does_not_hard_require_access(self) -> None:
+        project = tomllib.loads((ROOT / "pyproject.toml").read_text(encoding="utf-8"))["project"]
+        dependencies = tuple(project["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:
+        offenders: list[str] = []
+        for path in (ROOT / "src" / "govoplan_policy").rglob("*.py"):
+            source = path.read_text(encoding="utf-8")
+            if "govoplan_access" in source:
+                offenders.append(str(path.relative_to(ROOT)))
+
+        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

@@ -0,0 +1,27 @@
+{
+  "name": "@govoplan/policy-webui",
+  "version": "0.1.9",
+  "private": true,
+  "type": "module",
+  "main": "src/index.ts",
+  "module": "src/index.ts",
+  "types": "src/index.ts",
+  "exports": {
+    ".": {
+      "types": "./src/index.ts",
+      "import": "./src/index.ts"
+    }
+  },
+  "peerDependencies": {
+    "@govoplan/core-webui": "^0.1.9",
+    "lucide-react": "^1.23.0",
+    "react": "^19.0.0",
+    "react-dom": "^19.0.0",
+    "react-router-dom": "^7.1.1"
+  },
+  "peerDependenciesMeta": {
+    "@govoplan/core-webui": {
+      "optional": true
+    }
+  }
+}

webui/src/api/adminTargets.ts

@@ -0,0 +1,56 @@
+import { apiFetch, type ApiSettings, type DeltaDeletedItem } from "@govoplan/core-webui";
+
+export type RoleSummary = {
+  id: string;
+  slug: string;
+  name: string;
+  permissions: string[];
+};
+
+export type GroupSummary = {
+  id: string;
+  slug: string;
+  name: string;
+  description?: string | null;
+  is_active: boolean;
+  member_count: number;
+  member_ids: string[];
+  roles: RoleSummary[];
+};
+
+export type UserAdminItem = {
+  id: string;
+  account_id: string;
+  tenant_id: string;
+  email: string;
+  display_name?: string | null;
+  is_active: boolean;
+  groups: GroupSummary[];
+  roles: RoleSummary[];
+};
+
+type DeltaResponseFields = {
+  deleted: DeltaDeletedItem[];
+  watermark?: string | null;
+  has_more: boolean;
+  full: boolean;
+};
+
+export type UserListDeltaResponse = { users: UserAdminItem[] } & DeltaResponseFields;
+export type GroupListDeltaResponse = { groups: GroupSummary[] } & DeltaResponseFields;
+
+function deltaSuffix(options: { since?: string | null; limit?: number } = {}): string {
+  const params = new URLSearchParams();
+  if (options.since) params.set("since", options.since);
+  if (options.limit) params.set("limit", String(options.limit));
+  const suffix = params.toString();
+  return suffix ? `?${suffix}` : "";
+}
+
+export function fetchUsersDelta(settings: ApiSettings, options: { since?: string | null; limit?: number } = {}): Promise<UserListDeltaResponse> {
+  return apiFetch(settings, `/api/v1/admin/users/delta${deltaSuffix(options)}`);
+}
+
+export function fetchGroupsDelta(settings: ApiSettings, options: { since?: string | null; limit?: number } = {}): Promise<GroupListDeltaResponse> {
+  return apiFetch(settings, `/api/v1/admin/groups/delta${deltaSuffix(options)}`);
+}

webui/src/features/policy/RetentionPoliciesPanel.tsx

@@ -0,0 +1,226 @@
+import { useEffect, useRef, useState } from "react";
+import {
+  AdminPageLayout,
+  adminErrorMessage,
+  Button,
+  Card,
+  ConfirmDialog,
+  mergeDeltaRows,
+  RetentionPolicyScopeManager,
+  runRetentionPolicy,
+  useDeltaWatermarks,
+  type ApiSettings,
+  type DeltaDeletedItem,
+  type PrivacyRetentionPolicyScope,
+  type RetentionPolicyTargetOption,
+  type RetentionRunResponse
+} from "@govoplan/core-webui";
+import { fetchGroupsDelta, fetchUsersDelta, type GroupListDeltaResponse, type GroupSummary, type UserAdminItem, type UserListDeltaResponse } from "../../api/adminTargets";
+
+type Props = {
+  settings: ApiSettings;
+  scopeType: Extract<PrivacyRetentionPolicyScope, "system" | "tenant" | "user" | "group">;
+  canWrite: boolean;
+};
+
+type DeltaResponse = {
+  deleted: DeltaDeletedItem[];
+  watermark?: string | null;
+  has_more: boolean;
+  full: boolean;
+};
+
+const copy: Record<Props["scopeType"], { title: string; description: string; targetLabel?: string; policyTitle: string; policyDescription: string }> = {
+  system: {
+    title: "System retention",
+    description: "Instance-wide privacy retention policy and lower-level override limits.",
+    policyTitle: "System retention policy",
+    policyDescription: "Set concrete system retention values. Override switches define whether lower levels may narrow each value."
+  },
+  tenant: {
+    title: "Tenant retention",
+    description: "Tenant-level privacy and retention limits for the active tenant.",
+    policyTitle: "Tenant retention policy",
+    policyDescription: "Tenant limits may only narrow the system policy where the parent policy allows overrides."
+  },
+  user: {
+    title: "User retention",
+    description: "User-scoped retention limits for campaigns owned by a user.",
+    targetLabel: "User",
+    policyTitle: "User retention policy",
+    policyDescription: "User limits may only narrow inherited system and tenant policy."
+  },
+  group: {
+    title: "Group retention",
+    description: "Group-scoped retention limits for campaigns owned by a group.",
+    targetLabel: "Group",
+    policyTitle: "Group retention policy",
+    policyDescription: "Group limits may only narrow inherited system and tenant policy."
+  }
+};
+
+export default function RetentionPoliciesPanel({ settings, scopeType, canWrite }: Props) {
+  const [targets, setTargets] = useState<RetentionPolicyTargetOption[]>([]);
+  const usersRef = useRef<UserAdminItem[]>([]);
+  const groupsRef = useRef<GroupSummary[]>([]);
+  const { getDeltaWatermark, setDeltaWatermark, resetDeltaWatermark } = useDeltaWatermarks();
+  const [loadingTargets, setLoadingTargets] = useState(scopeType === "user" || scopeType === "group");
+  const [targetError, setTargetError] = useState("");
+  const [busy, setBusy] = useState(false);
+  const [success, setSuccess] = useState("");
+  const [runError, setRunError] = useState("");
+  const [confirmRetentionRun, setConfirmRetentionRun] = useState(false);
+  const [retentionResult, setRetentionResult] = useState<RetentionRunResponse | null>(null);
+
+  useEffect(() => {
+    usersRef.current = [];
+    groupsRef.current = [];
+    resetDeltaWatermark();
+    void loadTargets();
+  }, [settings.accessToken, settings.apiBaseUrl, settings.apiKey, scopeType, resetDeltaWatermark]);
+
+  async function loadTargets() {
+    if (scopeType !== "user" && scopeType !== "group") {
+      setTargets([]);
+      setLoadingTargets(false);
+      setTargetError("");
+      return;
+    }
+    setLoadingTargets(true);
+    setTargetError("");
+    try {
+      if (scopeType === "user") {
+        const users = await loadDeltaRows<UserAdminItem, UserListDeltaResponse>(
+          usersRef.current,
+          "policy:retention-users",
+          getDeltaWatermark,
+          setDeltaWatermark,
+          (since) => fetchUsersDelta(settings, { since }),
+          (response) => response.users,
+          (user) => user.id,
+          "access_user",
+          sortUsers
+        );
+        usersRef.current = users;
+        setTargets(users.map((user) => ({
+          id: user.id,
+          label: user.display_name || user.email,
+          secondary: user.display_name ? user.email : null
+        })));
+      } else {
+        const groups = await loadDeltaRows<GroupSummary, GroupListDeltaResponse>(
+          groupsRef.current,
+          "policy:retention-groups",
+          getDeltaWatermark,
+          setDeltaWatermark,
+          (since) => fetchGroupsDelta(settings, { since }),
+          (response) => response.groups,
+          (group) => group.id,
+          "access_group",
+          sortGroups
+        );
+        groupsRef.current = groups;
+        setTargets(groups.map((group) => ({
+          id: group.id,
+          label: group.name,
+          secondary: group.slug
+        })));
+      }
+    } catch (err) {
+      setTargets([]);
+      setTargetError(adminErrorMessage(err));
+    } finally {
+      setLoadingTargets(false);
+    }
+  }
+
+  async function runRetention(dryRun: boolean) {
+    setBusy(true);
+    setRunError("");
+    setSuccess("");
+    try {
+      const response = await runRetentionPolicy(settings, dryRun);
+      setRetentionResult(response);
+      setSuccess(dryRun ? "Retention dry run completed." : "Retention policy applied.");
+      setConfirmRetentionRun(false);
+    } catch (err) {
+      setRunError(adminErrorMessage(err));
+    } finally {
+      setBusy(false);
+    }
+  }
+
+  const labels = copy[scopeType];
+
+  return (
+    <>
+      <AdminPageLayout title={labels.title} description={labels.description} loading={loadingTargets} error={targetError || runError} success={success}>
+        <RetentionPolicyScopeManager
+          settings={settings}
+          scopeType={scopeType}
+          targetOptions={targets}
+          targetLabel={labels.targetLabel}
+          title={labels.policyTitle}
+          description={labels.policyDescription}
+          canWrite={canWrite}
+        />
+
+        {scopeType === "system" && (
+          <div className="retention-run-card">
+            <Card title="Retention execution">
+              <p className="muted small-note">Run the saved effective retention policy against retained platform data.</p>
+              <div className="button-row compact-actions subsection-bottom-actions">
+                <Button onClick={() => void runRetention(true)} disabled={!canWrite || busy}>Dry run</Button>
+                <Button variant="danger" onClick={() => setConfirmRetentionRun(true)} disabled={!canWrite || busy}>Apply retention</Button>
+              </div>
+              {retentionResult && <pre className="admin-json-preview">{JSON.stringify(retentionResult.result, null, 2)}</pre>}
+            </Card>
+          </div>
+        )}
+      </AdminPageLayout>
+      <ConfirmDialog
+        open={confirmRetentionRun}
+        title="Apply retention policy"
+        message="This will redact or delete eligible retained data according to the saved policy."
+        confirmLabel="Apply retention"
+        tone="danger"
+        busy={busy}
+        onCancel={() => setConfirmRetentionRun(false)}
+        onConfirm={() => void runRetention(false)}
+      />
+    </>
+  );
+}
+
+async function loadDeltaRows<TItem, TResponse extends DeltaResponse>(
+  current: TItem[],
+  key: string,
+  getDeltaWatermark: (key: string) => string | null,
+  setDeltaWatermark: (key: string, watermark: string | null | undefined) => void,
+  fetchDelta: (since: string | null) => Promise<TResponse>,
+  rowsFromResponse: (response: TResponse) => TItem[],
+  getKey: (item: TItem) => string,
+  deletedResourceType: string,
+  sort?: (left: TItem, right: TItem) => number
+): Promise<TItem[]> {
+  let nextWatermark = getDeltaWatermark(key);
+  let merged = current;
+  let hasMore = false;
+  do {
+    const response = await fetchDelta(nextWatermark);
+    const rows = rowsFromResponse(response);
+    merged = response.full ? rows : mergeDeltaRows(merged, rows, response.deleted, getKey, { deletedResourceType, sort });
+    nextWatermark = response.watermark ?? null;
+    hasMore = response.has_more;
+  } while (hasMore);
+  setDeltaWatermark(key, nextWatermark);
+  return merged;
+}
+
+function sortUsers(left: UserAdminItem, right: UserAdminItem): number {
+  return left.email.localeCompare(right.email);
+}
+
+function sortGroups(left: GroupSummary, right: GroupSummary): number {
+  return left.name.localeCompare(right.name) || left.slug.localeCompare(right.slug);
+}

webui/src/index.ts

@@ -0,0 +1,5 @@
+export { default } from "./module";
+export * from "./module";
+export * from "./api/adminTargets";
+export { default as RetentionPoliciesPanel } from "./features/policy/RetentionPoliciesPanel";
+export type { PlatformWebModule } from "@govoplan/core-webui";

webui/src/module.ts

@@ -0,0 +1,79 @@
+import { createElement, lazy } from "react";
+import { hasScope, type AdminSectionsUiCapability, type PlatformWebModule } from "@govoplan/core-webui";
+
+const RetentionPoliciesPanel = lazy(() => import("./features/policy/RetentionPoliciesPanel"));
+
+const policyAdminSections: AdminSectionsUiCapability = {
+  sections: [
+    {
+      id: "system-retention",
+      surfaceId: "policy.admin.system-retention",
+      label: "Retention",
+      group: "SYSTEM",
+      order: 80,
+      allOf: ["system:settings:read"],
+      render: ({ settings, auth }) => createElement(RetentionPoliciesPanel, {
+        settings,
+        scopeType: "system",
+        canWrite: hasScope(auth, "system:settings:write")
+      })
+    },
+    {
+      id: "tenant-retention",
+      surfaceId: "policy.admin.tenant-retention",
+      label: "Retention",
+      group: "TENANT",
+      order: 80,
+      allOf: ["admin:policies:read"],
+      render: ({ settings, auth }) => createElement(RetentionPoliciesPanel, {
+        settings,
+        scopeType: "tenant",
+        canWrite: hasScope(auth, "admin:policies:write")
+      })
+    },
+    {
+      id: "tenant-group-retention",
+      surfaceId: "policy.admin.group-retention",
+      label: "Retention",
+      group: "GROUP",
+      order: 30,
+      allOf: ["admin:policies:read", "admin:groups:read"],
+      render: ({ settings, auth }) => createElement(RetentionPoliciesPanel, {
+        settings,
+        scopeType: "group",
+        canWrite: hasScope(auth, "admin:policies:write")
+      })
+    },
+    {
+      id: "tenant-user-retention",
+      surfaceId: "policy.admin.user-retention",
+      label: "Retention",
+      group: "USER",
+      order: 30,
+      allOf: ["admin:policies:read", "admin:users:read"],
+      render: ({ settings, auth }) => createElement(RetentionPoliciesPanel, {
+        settings,
+        scopeType: "user",
+        canWrite: hasScope(auth, "admin:policies:write")
+      })
+    }
+  ]
+};
+
+export const policyModule: PlatformWebModule = {
+  id: "policy",
+  label: "Policy",
+  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
+  }
+};
+
+export default policyModule;