Persist hierarchical definition policy overrides

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

@@ -26,8 +26,18 @@ from govoplan_policy.backend.retention import (
     set_privacy_policy_for_scope,
     simulate_privacy_policy_change,
 )
+from govoplan_policy.backend.definition_policy_service import (
+    DefinitionPolicyError,
+    definition_policy_response_payload,
+    definition_policy_state,
+    remove_definition_policy,
+    save_definition_policy,
+)
+from govoplan_policy.backend.policy_overrides import PolicyOverrideError
 
 from .schemas import (
+    DefinitionPolicyScopeRequest,
+    DefinitionPolicyScopeResponse,
     PrivacyRetentionPolicyExplainResponse,
     PrivacyRetentionPolicyItem,
     PrivacyRetentionPolicyScopeRequest,
@@ -42,7 +52,9 @@ router = APIRouter(prefix="/admin", tags=["admin"])
 
 def _require_permission(principal: ApiPrincipal, scope: str) -> None:
     if not has_scope(principal, scope):
-        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=f"Missing scope: {scope}")
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN, detail=f"Missing scope: {scope}"
+        )
 
 
 def _require_privacy_policy_read(principal: ApiPrincipal, scope_type: str) -> None:
@@ -66,7 +78,237 @@ def _configuration_control_http_error(exc: ConfigurationControlError) -> HTTPExc
     )
 
 
-@router.get("/privacy-retention/policies/{scope_type}", response_model=PrivacyRetentionPolicyScopeResponse)
+def _definition_policy_response(
+    *,
+    module_id: str,
+    scope_type: str,
+    scope_id: str | None,
+    state,
+) -> DefinitionPolicyScopeResponse:
+    return DefinitionPolicyScopeResponse(
+        module_id=module_id,
+        scope_type=scope_type,
+        scope_id=scope_id,
+        **definition_policy_response_payload(state),
+    )
+
+
+@router.get(
+    "/definition-policies/{module_id}/{scope_type}",
+    response_model=DefinitionPolicyScopeResponse,
+)
+def read_definition_policy(
+    module_id: str,
+    scope_type: str,
+    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_read(principal, clean_scope)
+    try:
+        state = definition_policy_state(
+            session,
+            module_id=module_id,
+            tenant_id=principal.tenant_id,
+            scope_type=clean_scope,
+            scope_id=scope_id,
+        )
+        return _definition_policy_response(
+            module_id=module_id,
+            scope_type=clean_scope,
+            scope_id=scope_id,
+            state=state,
+        )
+    except (DefinitionPolicyError, PolicyOverrideError) as exc:
+        raise HTTPException(
+            status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
+            detail=str(exc),
+        ) from exc
+
+
+@router.put(
+    "/definition-policies/{module_id}/{scope_type}",
+    response_model=DefinitionPolicyScopeResponse,
+)
+def write_definition_policy(
+    module_id: str,
+    scope_type: str,
+    payload: DefinitionPolicyScopeRequest,
+    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)
+    policy_value = payload.policy.model_dump(mode="json", exclude_none=True)
+    try:
+        before = definition_policy_state(
+            session,
+            module_id=module_id,
+            tenant_id=principal.tenant_id,
+            scope_type=clean_scope,
+            scope_id=scope_id,
+        )
+        if clean_scope == "system":
+            approval = ensure_configuration_change_allowed(
+                session,
+                key="definition_policy",
+                value=policy_value,
+                actor_user_id=principal.user.id,
+                actor_scopes=tuple(principal.scopes),
+                change_request_id=payload.change_request_id,
+                target={"module_id": module_id, "scope_type": clean_scope},
+            )
+        else:
+            approval = None
+        state = save_definition_policy(
+            session,
+            module_id=module_id,
+            tenant_id=principal.tenant_id,
+            scope_type=clean_scope,
+            scope_id=scope_id,
+            policy=policy_value,
+            actor_id=principal.user.id,
+        )
+        if clean_scope == "system":
+            record_configuration_change_applied(
+                session,
+                key="definition_policy",
+                before_value=definition_policy_response_payload(before)["policy"],
+                after_value=policy_value,
+                actor_user_id=principal.user.id,
+                approval=approval,
+                target={"module_id": module_id, "scope_type": clean_scope},
+                audit_event="definition_policy.updated",
+            )
+        audit_from_principal(
+            session,
+            principal,
+            action="definition_policy.updated",
+            scope="system" if clean_scope == "system" else "tenant",
+            object_type="definition_policy",
+            object_id=f"{module_id}:{clean_scope}:{scope_id or ''}",
+            details={
+                "module_id": module_id,
+                "scope_type": clean_scope,
+                "scope_id": scope_id,
+                "fields": sorted(policy_value),
+            },
+        )
+        session.commit()
+        return _definition_policy_response(
+            module_id=module_id,
+            scope_type=clean_scope,
+            scope_id=scope_id,
+            state=state,
+        )
+    except ConfigurationControlError as exc:
+        session.rollback()
+        raise _configuration_control_http_error(exc) from exc
+    except (DefinitionPolicyError, PolicyOverrideError) as exc:
+        session.rollback()
+        raise HTTPException(
+            status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
+            detail=str(exc),
+        ) from exc
+
+
+@router.delete(
+    "/definition-policies/{module_id}/{scope_type}",
+    response_model=DefinitionPolicyScopeResponse,
+)
+def delete_definition_policy_route(
+    module_id: str,
+    scope_type: str,
+    scope_id: str | None = Query(default=None),
+    change_request_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:
+        before = definition_policy_state(
+            session,
+            module_id=module_id,
+            tenant_id=principal.tenant_id,
+            scope_type=clean_scope,
+            scope_id=scope_id,
+        )
+        if clean_scope == "system":
+            approval = ensure_configuration_change_allowed(
+                session,
+                key="definition_policy",
+                value={},
+                actor_user_id=principal.user.id,
+                actor_scopes=tuple(principal.scopes),
+                change_request_id=change_request_id,
+                target={"module_id": module_id, "scope_type": clean_scope},
+            )
+        else:
+            approval = None
+        removed = remove_definition_policy(
+            session,
+            module_id=module_id,
+            tenant_id=principal.tenant_id,
+            scope_type=clean_scope,
+            scope_id=scope_id,
+        )
+        if removed:
+            if clean_scope == "system":
+                record_configuration_change_applied(
+                    session,
+                    key="definition_policy",
+                    before_value=definition_policy_response_payload(before)["policy"],
+                    after_value={},
+                    actor_user_id=principal.user.id,
+                    approval=approval,
+                    target={"module_id": module_id, "scope_type": clean_scope},
+                    audit_event="definition_policy.removed",
+                )
+            audit_from_principal(
+                session,
+                principal,
+                action="definition_policy.removed",
+                scope="system" if clean_scope == "system" else "tenant",
+                object_type="definition_policy",
+                object_id=f"{module_id}:{clean_scope}:{scope_id or ''}",
+                details={
+                    "module_id": module_id,
+                    "scope_type": clean_scope,
+                    "scope_id": scope_id,
+                },
+            )
+        session.commit()
+        state = definition_policy_state(
+            session,
+            module_id=module_id,
+            tenant_id=principal.tenant_id,
+            scope_type=clean_scope,
+            scope_id=scope_id,
+        )
+        return _definition_policy_response(
+            module_id=module_id,
+            scope_type=clean_scope,
+            scope_id=scope_id,
+            state=state,
+        )
+    except ConfigurationControlError as exc:
+        session.rollback()
+        raise _configuration_control_http_error(exc) from exc
+    except (DefinitionPolicyError, PolicyOverrideError) as exc:
+        session.rollback()
+        raise HTTPException(
+            status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
+            detail=str(exc),
+        ) from exc
+
+
+@router.get(
+    "/privacy-retention/policies/{scope_type}",
+    response_model=PrivacyRetentionPolicyScopeResponse,
+)
 def read_privacy_retention_policy(
     scope_type: str,
     scope_id: str | None = Query(default=None),
@@ -76,23 +318,59 @@ def read_privacy_retention_policy(
     clean_scope = scope_type.strip().casefold()
     _require_privacy_policy_read(principal, clean_scope)
     try:
-        policy = get_privacy_policy_for_scope(session, tenant_id=principal.tenant_id, scope_type=clean_scope, scope_id=scope_id)
+        policy = get_privacy_policy_for_scope(
-        effective = _effective_privacy_policy_for_response(session, tenant_id=principal.tenant_id, scope_type=clean_scope, scope_id=scope_id)
+            session,
-        parent = _parent_privacy_policy_for_response(session, tenant_id=principal.tenant_id, scope_type=clean_scope, scope_id=scope_id)
+            tenant_id=principal.tenant_id,
+            scope_type=clean_scope,
+            scope_id=scope_id,
+        )
+        effective = _effective_privacy_policy_for_response(
+            session,
+            tenant_id=principal.tenant_id,
+            scope_type=clean_scope,
+            scope_id=scope_id,
+        )
+        parent = _parent_privacy_policy_for_response(
+            session,
+            tenant_id=principal.tenant_id,
+            scope_type=clean_scope,
+            scope_id=scope_id,
+        )
         return PrivacyRetentionPolicyScopeResponse(
             scope_type=clean_scope,
             scope_id=scope_id,
             policy=policy,
-            effective_policy=PrivacyRetentionPolicyItem.model_validate(effective.model_dump(mode="json")),
+            effective_policy=PrivacyRetentionPolicyItem.model_validate(
-            parent_policy=PrivacyRetentionPolicyItem.model_validate(parent.model_dump(mode="json")) if parent else None,
+                effective.model_dump(mode="json")
-            effective_policy_sources=_effective_privacy_policy_sources_for_response(session, tenant_id=principal.tenant_id, scope_type=clean_scope, scope_id=scope_id),
+            ),
-            parent_policy_sources=_parent_privacy_policy_sources_for_response(session, tenant_id=principal.tenant_id, scope_type=clean_scope, scope_id=scope_id),
+            parent_policy=PrivacyRetentionPolicyItem.model_validate(
+                parent.model_dump(mode="json")
+            )
+            if parent
+            else None,
+            effective_policy_sources=_effective_privacy_policy_sources_for_response(
+                session,
+                tenant_id=principal.tenant_id,
+                scope_type=clean_scope,
+                scope_id=scope_id,
+            ),
+            parent_policy_sources=_parent_privacy_policy_sources_for_response(
+                session,
+                tenant_id=principal.tenant_id,
+                scope_type=clean_scope,
+                scope_id=scope_id,
+            ),
         )
     except PrivacyPolicyError as exc:
-        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)
+        ) from exc
 
 
-@router.put("/privacy-retention/policies/{scope_type}", response_model=PrivacyRetentionPolicyScopeResponse)
+@router.put(
+    "/privacy-retention/policies/{scope_type}",
+    response_model=PrivacyRetentionPolicyScopeResponse,
+)
 def write_privacy_retention_policy(
     scope_type: str,
     payload: PrivacyRetentionPolicyScopeRequest,
@@ -105,7 +383,12 @@ def write_privacy_retention_policy(
     policy_value = payload.policy.model_dump(mode="json", exclude_none=True)
     before_value: dict[str, Any] | None = None
     if clean_scope == "system":
-        before_value = get_privacy_policy_for_scope(session, tenant_id=principal.tenant_id, scope_type=clean_scope, scope_id=scope_id)
+        before_value = get_privacy_policy_for_scope(
+            session,
+            tenant_id=principal.tenant_id,
+            scope_type=clean_scope,
+            scope_id=scope_id,
+        )
         try:
             approval = ensure_configuration_change_allowed(
                 session,
@@ -149,23 +432,54 @@ def write_privacy_retention_policy(
             details={"scope_type": clean_scope, "scope_id": scope_id},
         )
         session.commit()
-        effective = _effective_privacy_policy_for_response(session, tenant_id=principal.tenant_id, scope_type=clean_scope, scope_id=scope_id)
+        effective = _effective_privacy_policy_for_response(
-        parent = _parent_privacy_policy_for_response(session, tenant_id=principal.tenant_id, scope_type=clean_scope, scope_id=scope_id)
+            session,
+            tenant_id=principal.tenant_id,
+            scope_type=clean_scope,
+            scope_id=scope_id,
+        )
+        parent = _parent_privacy_policy_for_response(
+            session,
+            tenant_id=principal.tenant_id,
+            scope_type=clean_scope,
+            scope_id=scope_id,
+        )
         return PrivacyRetentionPolicyScopeResponse(
             scope_type=clean_scope,
             scope_id=scope_id,
             policy=policy,
-            effective_policy=PrivacyRetentionPolicyItem.model_validate(effective.model_dump(mode="json")),
+            effective_policy=PrivacyRetentionPolicyItem.model_validate(
-            parent_policy=PrivacyRetentionPolicyItem.model_validate(parent.model_dump(mode="json")) if parent else None,
+                effective.model_dump(mode="json")
-            effective_policy_sources=_effective_privacy_policy_sources_for_response(session, tenant_id=principal.tenant_id, scope_type=clean_scope, scope_id=scope_id),
+            ),
-            parent_policy_sources=_parent_privacy_policy_sources_for_response(session, tenant_id=principal.tenant_id, scope_type=clean_scope, scope_id=scope_id),
+            parent_policy=PrivacyRetentionPolicyItem.model_validate(
+                parent.model_dump(mode="json")
+            )
+            if parent
+            else None,
+            effective_policy_sources=_effective_privacy_policy_sources_for_response(
+                session,
+                tenant_id=principal.tenant_id,
+                scope_type=clean_scope,
+                scope_id=scope_id,
+            ),
+            parent_policy_sources=_parent_privacy_policy_sources_for_response(
+                session,
+                tenant_id=principal.tenant_id,
+                scope_type=clean_scope,
+                scope_id=scope_id,
+            ),
         )
     except PrivacyPolicyError as exc:
         session.rollback()
-        raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail=str(exc)) from exc
+        raise HTTPException(
+            status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail=str(exc)
+        ) from exc
 
 
-@router.get("/privacy-retention/policies/{scope_type}/explain", response_model=PrivacyRetentionPolicyExplainResponse)
+@router.get(
+    "/privacy-retention/policies/{scope_type}/explain",
+    response_model=PrivacyRetentionPolicyExplainResponse,
+)
 def explain_privacy_retention_policy(
     scope_type: str,
     scope_id: str | None = Query(default=None),
@@ -175,16 +489,40 @@ def explain_privacy_retention_policy(
     clean_scope = scope_type.strip().casefold()
     _require_privacy_policy_read(principal, clean_scope)
     try:
-        effective = _effective_privacy_policy_for_response(session, tenant_id=principal.tenant_id, scope_type=clean_scope, scope_id=scope_id)
+        effective = _effective_privacy_policy_for_response(
-        parent = _parent_privacy_policy_for_response(session, tenant_id=principal.tenant_id, scope_type=clean_scope, scope_id=scope_id)
+            session,
-        effective_sources = _effective_privacy_policy_sources_for_response(session, tenant_id=principal.tenant_id, scope_type=clean_scope, scope_id=scope_id)
+            tenant_id=principal.tenant_id,
-        parent_sources = _parent_privacy_policy_sources_for_response(session, tenant_id=principal.tenant_id, scope_type=clean_scope, scope_id=scope_id)
+            scope_type=clean_scope,
+            scope_id=scope_id,
+        )
+        parent = _parent_privacy_policy_for_response(
+            session,
+            tenant_id=principal.tenant_id,
+            scope_type=clean_scope,
+            scope_id=scope_id,
+        )
+        effective_sources = _effective_privacy_policy_sources_for_response(
+            session,
+            tenant_id=principal.tenant_id,
+            scope_type=clean_scope,
+            scope_id=scope_id,
+        )
+        parent_sources = _parent_privacy_policy_sources_for_response(
+            session,
+            tenant_id=principal.tenant_id,
+            scope_type=clean_scope,
+            scope_id=scope_id,
+        )
         blocked_fields = _blocked_privacy_retention_fields(parent)
         decision_sources = parent_sources or effective_sources
         decision = PolicyDecision(
             allowed=not blocked_fields,
-            reason="Parent retention policy locks lower-level changes." if blocked_fields else None,
+            reason="Parent retention policy locks lower-level changes."
-            source_path=tuple(PolicySourceStep.from_mapping(source) for source in decision_sources),
+            if blocked_fields
+            else None,
+            source_path=tuple(
+                PolicySourceStep.from_mapping(source) for source in decision_sources
+            ),
             requirements=tuple(blocked_fields),
             details={"blocked_fields": blocked_fields},
         )
@@ -192,17 +530,28 @@ def explain_privacy_retention_policy(
             scope_type=clean_scope,
             scope_id=scope_id,
             decision=decision.to_dict(),
-            effective_policy=PrivacyRetentionPolicyItem.model_validate(effective.model_dump(mode="json")),
+            effective_policy=PrivacyRetentionPolicyItem.model_validate(
-            parent_policy=PrivacyRetentionPolicyItem.model_validate(parent.model_dump(mode="json")) if parent else None,
+                effective.model_dump(mode="json")
+            ),
+            parent_policy=PrivacyRetentionPolicyItem.model_validate(
+                parent.model_dump(mode="json")
+            )
+            if parent
+            else None,
             effective_policy_sources=effective_sources,
             parent_policy_sources=parent_sources,
             blocked_fields=blocked_fields,
         )
     except PrivacyPolicyError as exc:
-        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc
+        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)
+@router.post(
+    "/privacy-retention/policies/{scope_type}/simulate",
+    response_model=PrivacyRetentionPolicySimulationResponse,
+)
 def simulate_privacy_retention_policy(
     scope_type: str,
     payload: PrivacyRetentionPolicyScopeRequest,
@@ -225,7 +574,9 @@ def simulate_privacy_retention_policy(
             ),
         )
     except PrivacyPolicyError as exc:
-        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)
+        ) from exc
 
 
 def _blocked_privacy_retention_fields(parent) -> list[str]:
@@ -233,36 +584,64 @@ def _blocked_privacy_retention_fields(parent) -> list[str]:
         return []
     payload = parent.model_dump(mode="json")
     allow_lower_level_limits = payload.get("allow_lower_level_limits") or {}
-    return [key for key in RETENTION_POLICY_FIELD_KEYS if allow_lower_level_limits.get(key) is False]
+    return [
+        key
+        for key in RETENTION_POLICY_FIELD_KEYS
+        if allow_lower_level_limits.get(key) is False
+    ]
 
 
-def _parent_privacy_policy_sources_for_response(session: Session, *, tenant_id: str, scope_type: str, scope_id: str | None):
+def _parent_privacy_policy_sources_for_response(
+    session: Session, *, tenant_id: str, scope_type: str, scope_id: str | None
+):
     if scope_type == "system":
         return []
-    return parent_privacy_policy_sources(session, tenant_id=tenant_id, scope_type=scope_type, scope_id=scope_id or (tenant_id if scope_type == "tenant" else None))
+    return parent_privacy_policy_sources(
+        session,
+        tenant_id=tenant_id,
+        scope_type=scope_type,
+        scope_id=scope_id or (tenant_id if scope_type == "tenant" else None),
+    )
 
 
-def _effective_privacy_policy_sources_for_response(session: Session, *, tenant_id: str, scope_type: str, scope_id: str | None):
+def _effective_privacy_policy_sources_for_response(
+    session: Session, *, tenant_id: str, scope_type: str, scope_id: str | None
+):
     if scope_type == "system":
         return effective_privacy_policy_sources(session)
     if scope_type == "tenant":
-        return effective_privacy_policy_sources(session, tenant_id=scope_id or tenant_id)
+        return effective_privacy_policy_sources(
+            session, tenant_id=scope_id or tenant_id
+        )
     if scope_type == "campaign" and scope_id:
         return effective_privacy_policy_sources(session, campaign_id=scope_id)
     if scope_type == "user" and scope_id:
-        return effective_privacy_policy_sources(session, tenant_id=tenant_id, owner_user_id=scope_id)
+        return effective_privacy_policy_sources(
+            session, tenant_id=tenant_id, owner_user_id=scope_id
+        )
     if scope_type == "group" and scope_id:
-        return effective_privacy_policy_sources(session, tenant_id=tenant_id, owner_group_id=scope_id)
+        return effective_privacy_policy_sources(
+            session, tenant_id=tenant_id, owner_group_id=scope_id
+        )
     return effective_privacy_policy_sources(session, tenant_id=tenant_id)
 
 
-def _parent_privacy_policy_for_response(session: Session, *, tenant_id: str, scope_type: str, scope_id: str | None):
+def _parent_privacy_policy_for_response(
+    session: Session, *, tenant_id: str, scope_type: str, scope_id: str | None
+):
     if scope_type == "system":
         return None
-    return parent_privacy_policy(session, tenant_id=tenant_id, scope_type=scope_type, scope_id=scope_id or (tenant_id if scope_type == "tenant" else None))
+    return parent_privacy_policy(
+        session,
+        tenant_id=tenant_id,
+        scope_type=scope_type,
+        scope_id=scope_id or (tenant_id if scope_type == "tenant" else None),
+    )
 
 
-def _effective_privacy_policy_for_response(session: Session, *, tenant_id: str, scope_type: str, scope_id: str | None):
+def _effective_privacy_policy_for_response(
+    session: Session, *, tenant_id: str, scope_type: str, scope_id: str | None
+):
     if scope_type == "system":
         return effective_privacy_policy(session)
     if scope_type == "tenant":
@@ -270,9 +649,13 @@ def _effective_privacy_policy_for_response(session: Session, *, tenant_id: str,
     if scope_type == "campaign" and scope_id:
         return effective_privacy_policy(session, campaign_id=scope_id)
     if scope_type == "user" and scope_id:
-        return effective_privacy_policy(session, tenant_id=tenant_id, owner_user_id=scope_id)
+        return effective_privacy_policy(
+            session, tenant_id=tenant_id, owner_user_id=scope_id
+        )
     if scope_type == "group" and scope_id:
-        return effective_privacy_policy(session, tenant_id=tenant_id, owner_group_id=scope_id)
+        return effective_privacy_policy(
+            session, tenant_id=tenant_id, owner_group_id=scope_id
+        )
     return effective_privacy_policy(session, tenant_id=tenant_id)
 
 

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

@@ -4,7 +4,10 @@ from typing import Any, Literal
 
 from pydantic import BaseModel, ConfigDict, Field
 
-from govoplan_core.privacy.schemas import PrivacyRetentionPolicyItem, PrivacyRetentionPolicyPatchItem
+from govoplan_core.privacy.schemas import (
+    PrivacyRetentionPolicyItem,
+    PrivacyRetentionPolicyPatchItem,
+)
 
 
 class PolicySourceStepItem(BaseModel):
@@ -19,7 +22,9 @@ class PolicySourceStepItem(BaseModel):
 class PrivacyRetentionPolicyScopeRequest(BaseModel):
     model_config = ConfigDict(extra="forbid")
 
-    policy: PrivacyRetentionPolicyPatchItem = Field(default_factory=PrivacyRetentionPolicyPatchItem)
+    policy: PrivacyRetentionPolicyPatchItem = Field(
+        default_factory=PrivacyRetentionPolicyPatchItem
+    )
     change_request_id: str | None = None
 
 
@@ -41,6 +46,37 @@ class PolicyDecisionItem(BaseModel):
     details: dict[str, Any] = Field(default_factory=dict)
 
 
+class DefinitionPolicyItem(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    allow_view: bool | None = None
+    allow_edit: bool | None = None
+    inherit_to_lower_scopes: bool | None = None
+    allow_run: bool | None = None
+    allow_reuse: bool | None = None
+    allow_automation: bool | None = None
+
+
+class DefinitionPolicyScopeRequest(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    policy: DefinitionPolicyItem = Field(default_factory=DefinitionPolicyItem)
+    change_request_id: str | None = None
+
+
+class DefinitionPolicyScopeResponse(BaseModel):
+    module_id: str
+    scope_type: Literal["system", "tenant", "group", "user"]
+    scope_id: str | None = None
+    id: str | None = None
+    revision: int | None = None
+    policy: dict[str, Any] = Field(default_factory=dict)
+    effective_policy: dict[str, bool] = Field(default_factory=dict)
+    parent_policy: dict[str, bool] = Field(default_factory=dict)
+    source_path: list[PolicySourceStepItem] = Field(default_factory=list)
+    diagnostics: list[dict[str, str]] = Field(default_factory=list)
+
+
 class PrivacyRetentionPolicyExplainResponse(BaseModel):
     scope_type: Literal["system", "tenant", "user", "group", "campaign"]
     scope_id: str | None = None

src/govoplan_policy/backend/db/__init__.py

@@ -0,0 +1,5 @@
+"""Policy-owned persistence models."""
+
+from govoplan_policy.backend.db.models import PolicyOverride
+
+__all__ = ["PolicyOverride"]

src/govoplan_policy/backend/db/models.py

@@ -0,0 +1,52 @@
+from __future__ import annotations
+
+import uuid
+from typing import Any
+
+from sqlalchemy import Index, Integer, JSON, String, UniqueConstraint
+from sqlalchemy.orm import Mapped, mapped_column
+
+from govoplan_core.db.base import Base, TimestampMixin
+
+
+class PolicyOverride(Base, TimestampMixin):
+    __tablename__ = "policy_overrides"
+    __table_args__ = (
+        UniqueConstraint(
+            "policy_family",
+            "target_key",
+            "scope_key",
+            name="uq_policy_override_family_target_scope",
+        ),
+        Index(
+            "ix_policy_overrides_resolution",
+            "policy_family",
+            "target_key",
+            "tenant_id",
+            "scope_type",
+            "scope_id",
+        ),
+    )
+
+    id: Mapped[str] = mapped_column(
+        String(36),
+        primary_key=True,
+        default=lambda: str(uuid.uuid4()),
+    )
+    policy_family: Mapped[str] = mapped_column(String(40), nullable=False)
+    target_key: Mapped[str] = mapped_column(String(120), nullable=False)
+    tenant_id: Mapped[str | None] = mapped_column(String(36), nullable=True, index=True)
+    scope_type: Mapped[str] = mapped_column(String(20), nullable=False)
+    scope_id: Mapped[str | None] = mapped_column(String(255), nullable=True, index=True)
+    scope_key: Mapped[str] = mapped_column(String(320), nullable=False)
+    policy: Mapped[Any] = mapped_column(JSON, default=dict, nullable=False)
+    revision: Mapped[int] = mapped_column(Integer, default=1, nullable=False)
+    created_by: Mapped[str | None] = mapped_column(
+        String(255), nullable=True, index=True
+    )
+    updated_by: Mapped[str | None] = mapped_column(
+        String(255), nullable=True, index=True
+    )
+
+
+__all__ = ["PolicyOverride"]

src/govoplan_policy/backend/definition_governance.py

@@ -1,14 +1,36 @@
 from __future__ import annotations
 
 from collections.abc import Mapping
+from dataclasses import dataclass
 from typing import Any
 
+from sqlalchemy.orm import Session
+
 from govoplan_core.core.policy import (
     DefinitionGovernanceRequest,
     PolicyDecision,
     PolicySourceStep,
 )
 from govoplan_core.security.module_permissions import scopes_grant_compatible
+from govoplan_core.db.session import get_database
+from govoplan_policy.backend.policy_overrides import resolution_policy_overrides
+
+
+DEFINITION_POLICY_FIELDS = (
+    "allow_view",
+    "allow_edit",
+    "inherit_to_lower_scopes",
+    "allow_run",
+    "allow_reuse",
+    "allow_automation",
+)
+
+
+@dataclass(frozen=True, slots=True)
+class DefinitionPolicyResolution:
+    limits: Mapping[str, bool]
+    source_path: tuple[PolicySourceStep, ...] = ()
+    diagnostics: tuple[Mapping[str, str], ...] = ()
 
 
 class DefinitionGovernancePolicyProvider:
@@ -16,11 +38,19 @@ class DefinitionGovernancePolicyProvider:
 
     def resolve_definition_action(
         self,
+        session: object | None = None,
         *,
         request: DefinitionGovernanceRequest,
     ) -> PolicyDecision:
-        source_path = _source_path(request)
+        explicit_policy = _explicit_policy_resolution(session, request)
-        effective = _effective_limits(request)
+        source_path = _source_path(
+            request,
+            policy_sources=explicit_policy.source_path,
+        )
+        effective = _effective_limits(
+            request,
+            policy_limits=explicit_policy.limits,
+        )
         visible, visibility_reason = _visible(
             request,
             effective=effective,
@@ -34,16 +64,21 @@ class DefinitionGovernancePolicyProvider:
                 source_path=source_path,
                 request=request,
                 effective=effective,
+                policy_diagnostics=explicit_policy.diagnostics,
             )
 
         if action == "edit":
             editable, reason = _editable(request)
+            if editable and not effective["allow_edit"]:
+                editable = False
+                reason = "Editing is disabled by explicit Policy restrictions."
             return _decision(
                 editable,
                 reason,
                 source_path=source_path,
                 request=request,
                 effective=effective,
+                policy_diagnostics=explicit_policy.diagnostics,
             )
 
         if not visible:
@@ -53,6 +88,7 @@ class DefinitionGovernancePolicyProvider:
                 source_path=source_path,
                 request=request,
                 effective=effective,
+                policy_diagnostics=explicit_policy.diagnostics,
             )
 
         if action == "run":
@@ -63,6 +99,7 @@ class DefinitionGovernancePolicyProvider:
                     source_path=source_path,
                     request=request,
                     effective=effective,
+                    policy_diagnostics=explicit_policy.diagnostics,
                 )
             if request.status != "active":
                 return _decision(
@@ -71,6 +108,7 @@ class DefinitionGovernancePolicyProvider:
                     source_path=source_path,
                     request=request,
                     effective=effective,
+                    policy_diagnostics=explicit_policy.diagnostics,
                 )
             return _decision(
                 effective["allow_run"],
@@ -82,6 +120,7 @@ class DefinitionGovernancePolicyProvider:
                 source_path=source_path,
                 request=request,
                 effective=effective,
+                policy_diagnostics=explicit_policy.diagnostics,
             )
 
         if action in {"reuse", "derive"}:
@@ -95,6 +134,7 @@ class DefinitionGovernancePolicyProvider:
                 source_path=source_path,
                 request=request,
                 effective=effective,
+                policy_diagnostics=explicit_policy.diagnostics,
             )
 
         if action == "automate":
@@ -105,6 +145,7 @@ class DefinitionGovernancePolicyProvider:
                     source_path=source_path,
                     request=request,
                     effective=effective,
+                    policy_diagnostics=explicit_policy.diagnostics,
                 )
             if request.status != "active":
                 return _decision(
@@ -113,6 +154,7 @@ class DefinitionGovernancePolicyProvider:
                     source_path=source_path,
                     request=request,
                     effective=effective,
+                    policy_diagnostics=explicit_policy.diagnostics,
                 )
             allowed = effective["allow_run"] and effective["allow_automation"]
             return _decision(
@@ -125,6 +167,7 @@ class DefinitionGovernancePolicyProvider:
                 source_path=source_path,
                 request=request,
                 effective=effective,
+                policy_diagnostics=explicit_policy.diagnostics,
             )
 
         return _decision(
@@ -133,6 +176,7 @@ class DefinitionGovernancePolicyProvider:
             source_path=source_path,
             request=request,
             effective=effective,
+            policy_diagnostics=explicit_policy.diagnostics,
         )
 
 
@@ -141,6 +185,8 @@ def _visible(
     *,
     effective: Mapping[str, bool],
 ) -> tuple[bool, str | None]:
+    if not effective["allow_view"]:
+        return False, "Visibility is disabled by explicit Policy restrictions."
     scope = request.definition_scope
     actor = request.actor
     if scope.scope_type == "system":
@@ -176,36 +222,56 @@ def _editable(request: DefinitionGovernanceRequest) -> tuple[bool, str | None]:
         allowed = _has_scope(actor.scopes, "system:governance:write")
         return (
             allowed,
-            None if allowed else "System definitions require system governance permission.",
+            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."
+        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."
+        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 (
+            allowed,
+            None if allowed else "Definitions from another user are read-only.",
+        )
     return False, "The definition scope is invalid."
 
 
 def _effective_limits(
     request: DefinitionGovernanceRequest,
+    *,
+    policy_limits: Mapping[str, bool] | None = None,
 ) -> dict[str, bool]:
     ancestor = request.context.get("ancestor_limits")
     ancestor_limits = ancestor if isinstance(ancestor, Mapping) else {}
+    explicit = policy_limits or {}
     return {
+        "allow_view": _policy_flag(explicit, "allow_view"),
+        "allow_edit": _policy_flag(explicit, "allow_edit"),
         "inherit_to_lower_scopes": request.inherit_to_lower_scopes
-        and _ancestor_flag(ancestor_limits, "inherit_to_lower_scopes"),
+        and _ancestor_flag(ancestor_limits, "inherit_to_lower_scopes")
+        and _policy_flag(explicit, "inherit_to_lower_scopes"),
         "allow_run": request.allow_run
-        and _ancestor_flag(ancestor_limits, "allow_run"),
+        and _ancestor_flag(ancestor_limits, "allow_run")
+        and _policy_flag(explicit, "allow_run"),
         "allow_reuse": request.allow_reuse
-        and _ancestor_flag(ancestor_limits, "allow_reuse"),
+        and _ancestor_flag(ancestor_limits, "allow_reuse")
+        and _policy_flag(explicit, "allow_reuse"),
         "allow_automation": request.allow_automation
-        and _ancestor_flag(ancestor_limits, "allow_automation"),
+        and _ancestor_flag(ancestor_limits, "allow_automation")
+        and _policy_flag(explicit, "allow_automation"),
     }
 
 
@@ -214,10 +280,17 @@ def _ancestor_flag(value: Mapping[str, Any], key: str) -> bool:
     return True if raw is None else raw is True
 
 
+def _policy_flag(value: Mapping[str, bool], key: str) -> bool:
+    raw = value.get(key)
+    return True if raw is None else raw is True
+
+
 def _source_path(
     request: DefinitionGovernanceRequest,
+    *,
+    policy_sources: tuple[PolicySourceStep, ...] = (),
 ) -> tuple[PolicySourceStep, ...]:
-    steps: list[PolicySourceStep] = []
+    steps: list[PolicySourceStep] = list(policy_sources)
     ancestor = request.context.get("ancestor_limits")
     if isinstance(ancestor, Mapping):
         source = request.context.get("ancestor_source")
@@ -278,15 +351,14 @@ def _decision(
     source_path: tuple[PolicySourceStep, ...],
     request: DefinitionGovernanceRequest,
     effective: Mapping[str, bool],
+    policy_diagnostics: tuple[Mapping[str, str], ...] = (),
 ) -> PolicyDecision:
     return PolicyDecision(
         allowed=allowed,
         reason=reason,
         source_path=source_path,
         requirements=(
-            ()
+            () if allowed else (f"{request.module_id}.definition.{request.action}",)
-            if allowed
-            else (f"{request.module_id}.definition.{request.action}",)
         ),
         details={
             "module_id": request.module_id,
@@ -296,12 +368,117 @@ def _decision(
             "target_scope": request.target_scope.path,
             "definition_kind": request.definition_kind,
             "effective_limits": dict(effective),
+            "policy_diagnostics": [dict(item) for item in policy_diagnostics],
         },
     )
 
 
+def _explicit_policy_resolution(
+    session: object | None,
+    request: DefinitionGovernanceRequest,
+) -> DefinitionPolicyResolution:
+    if not isinstance(session, Session):
+        try:
+            with get_database().SessionLocal() as policy_session:
+                return _explicit_policy_resolution(policy_session, request)
+        except RuntimeError:
+            # Contract-only tests and offline tooling may resolve decisions
+            # without configuring a database. No persisted override exists in
+            # that context.
+            return DefinitionPolicyResolution(limits={})
+    cache_key = (
+        "definition",
+        request.module_id,
+        request.tenant_id,
+        tuple(sorted(request.actor.group_ids)),
+        request.actor.account_id,
+        request.actor.membership_id,
+    )
+    cache = session.info.setdefault("govoplan_policy_override_resolution", {})
+    if isinstance(cache, dict) and cache_key in cache:
+        cached = cache[cache_key]
+        if isinstance(cached, DefinitionPolicyResolution):
+            return cached
+    rows = resolution_policy_overrides(
+        session,
+        policy_family="definition",
+        target_keys=("*", request.module_id),
+        tenant_id=request.tenant_id,
+        group_ids=request.actor.group_ids,
+        user_ids=(request.actor.account_id, request.actor.membership_id or ""),
+    )
+    result = resolve_definition_policy_rows(rows)
+    if isinstance(cache, dict):
+        cache[cache_key] = result
+    return result
+
+
+def resolve_definition_policy_rows(
+    rows: object,
+) -> DefinitionPolicyResolution:
+    limits = {field: True for field in DEFINITION_POLICY_FIELDS}
+    steps: list[PolicySourceStep] = []
+    diagnostics: list[Mapping[str, str]] = []
+    for row in rows if isinstance(rows, (list, tuple)) else ():
+        policy, malformed = validate_definition_policy(row.policy)
+        if malformed:
+            limits.update({field: False for field in DEFINITION_POLICY_FIELDS})
+            applied_fields = DEFINITION_POLICY_FIELDS
+            source_policy: Mapping[str, Any] = {
+                "configuration_status": "invalid_fail_closed",
+                "target_key": row.target_key,
+            }
+            diagnostics.append(
+                {
+                    "code": "definition_policy.invalid",
+                    "scope": row.scope_key,
+                    "target_key": row.target_key,
+                }
+            )
+        else:
+            for field, value in policy.items():
+                limits[field] = limits[field] and value
+            applied_fields = tuple(sorted(policy))
+            source_policy = {**policy, "target_key": row.target_key}
+        steps.append(
+            PolicySourceStep(
+                scope_type=row.scope_type,  # type: ignore[arg-type]
+                scope_id=row.scope_id,
+                label=(f"{row.scope_type.capitalize()} definition policy"),
+                applied_fields=tuple(applied_fields),
+                policy=source_policy,
+            )
+        )
+    return DefinitionPolicyResolution(
+        limits=limits,
+        source_path=tuple(steps),
+        diagnostics=tuple(diagnostics),
+    )
+
+
+def validate_definition_policy(
+    value: object,
+) -> tuple[dict[str, bool], bool]:
+    if not isinstance(value, Mapping):
+        return {}, True
+    if any(str(key) not in DEFINITION_POLICY_FIELDS for key in value):
+        return {}, True
+    policy: dict[str, bool] = {}
+    for key, raw in value.items():
+        if not isinstance(raw, bool):
+            return {}, True
+        policy[str(key)] = raw
+    return policy, False
+
+
 def _has_scope(scopes: object, required: str) -> bool:
     return scopes_grant_compatible(scopes, required)  # type: ignore[arg-type]
 
 
-__all__ = ["DefinitionGovernancePolicyProvider"]
+__all__ = [
+    "DEFINITION_POLICY_FIELDS",
+    "DefinitionGovernancePolicyProvider",
+    "DefinitionPolicyResolution",
+    "resolve_definition_policy_rows",
+    "validate_definition_policy",
+]

src/govoplan_policy/backend/definition_policy_service.py

@@ -0,0 +1,219 @@
+from __future__ import annotations
+
+from collections.abc import Mapping
+from dataclasses import dataclass
+from typing import Any
+
+from sqlalchemy.orm import Session
+
+from govoplan_policy.backend.db.models import PolicyOverride
+from govoplan_policy.backend.definition_governance import (
+    DEFINITION_POLICY_FIELDS,
+    DefinitionPolicyResolution,
+    resolve_definition_policy_rows,
+    validate_definition_policy,
+)
+from govoplan_policy.backend.policy_overrides import (
+    delete_policy_override,
+    get_policy_override,
+    normalize_policy_scope,
+    resolution_policy_overrides,
+    set_policy_override,
+)
+
+
+class DefinitionPolicyError(ValueError):
+    pass
+
+
+@dataclass(frozen=True, slots=True)
+class DefinitionPolicyState:
+    row: PolicyOverride | None
+    local_policy: Mapping[str, bool]
+    effective: DefinitionPolicyResolution
+    parent: DefinitionPolicyResolution
+
+
+def definition_policy_state(
+    session: Session,
+    *,
+    module_id: str,
+    tenant_id: str,
+    scope_type: str,
+    scope_id: str | None = None,
+) -> DefinitionPolicyState:
+    _, clean_scope_id, _ = normalize_policy_scope(
+        tenant_id=tenant_id,
+        scope_type=scope_type,
+        scope_id=scope_id,
+    )
+    row = get_policy_override(
+        session,
+        policy_family="definition",
+        target_key=module_id,
+        tenant_id=tenant_id,
+        scope_type=scope_type,
+        scope_id=clean_scope_id,
+    )
+    local_policy, malformed = validate_definition_policy(
+        row.policy if row is not None else {}
+    )
+    if malformed:
+        local_policy = {}
+    rows = _rows_for_scope(
+        session,
+        module_id=module_id,
+        tenant_id=tenant_id,
+        scope_type=scope_type,
+        scope_id=clean_scope_id,
+    )
+    rank = _scope_rank(scope_type)
+    return DefinitionPolicyState(
+        row=row,
+        local_policy=local_policy,
+        effective=resolve_definition_policy_rows(rows),
+        parent=resolve_definition_policy_rows(
+            tuple(item for item in rows if _scope_rank(item.scope_type) < rank)
+        ),
+    )
+
+
+def save_definition_policy(
+    session: Session,
+    *,
+    module_id: str,
+    tenant_id: str,
+    scope_type: str,
+    scope_id: str | None,
+    policy: object,
+    actor_id: str | None,
+) -> DefinitionPolicyState:
+    clean_policy, malformed = validate_definition_policy(policy)
+    if malformed:
+        raise DefinitionPolicyError(
+            "Definition policy fields must be known boolean values"
+        )
+    before = definition_policy_state(
+        session,
+        module_id=module_id,
+        tenant_id=tenant_id,
+        scope_type=scope_type,
+        scope_id=scope_id,
+    )
+    broadened = sorted(
+        field
+        for field, value in clean_policy.items()
+        if value and before.parent.limits.get(field) is False
+    )
+    if broadened:
+        raise DefinitionPolicyError(
+            "Lower-scope policy cannot broaden parent restrictions: "
+            + ", ".join(broadened)
+        )
+    set_policy_override(
+        session,
+        policy_family="definition",
+        target_key=module_id,
+        tenant_id=tenant_id,
+        scope_type=scope_type,
+        scope_id=scope_id,
+        policy=clean_policy,
+        actor_id=actor_id,
+    )
+    _clear_resolution_cache(session)
+    return definition_policy_state(
+        session,
+        module_id=module_id,
+        tenant_id=tenant_id,
+        scope_type=scope_type,
+        scope_id=scope_id,
+    )
+
+
+def remove_definition_policy(
+    session: Session,
+    *,
+    module_id: str,
+    tenant_id: str,
+    scope_type: str,
+    scope_id: str | None,
+) -> bool:
+    row = get_policy_override(
+        session,
+        policy_family="definition",
+        target_key=module_id,
+        tenant_id=tenant_id,
+        scope_type=scope_type,
+        scope_id=scope_id,
+    )
+    if row is None:
+        return False
+    delete_policy_override(session, row)
+    _clear_resolution_cache(session)
+    return True
+
+
+def _rows_for_scope(
+    session: Session,
+    *,
+    module_id: str,
+    tenant_id: str,
+    scope_type: str,
+    scope_id: str | None,
+) -> tuple[PolicyOverride, ...]:
+    clean_scope = scope_type.strip().casefold()
+    rows = resolution_policy_overrides(
+        session,
+        policy_family="definition",
+        target_keys=("*", module_id),
+        tenant_id=tenant_id,
+        group_ids=(scope_id,) if clean_scope == "group" and scope_id else (),
+        user_ids=(scope_id,) if clean_scope == "user" and scope_id else (),
+    )
+    maximum_rank = _scope_rank(clean_scope)
+    return tuple(row for row in rows if _scope_rank(row.scope_type) <= maximum_rank)
+
+
+def _scope_rank(scope_type: str) -> int:
+    try:
+        return ("system", "tenant", "group", "user").index(
+            scope_type.strip().casefold()
+        )
+    except ValueError as exc:
+        raise DefinitionPolicyError(
+            "Definition policy scope must be system, tenant, group, or user"
+        ) from exc
+
+
+def _clear_resolution_cache(session: Session) -> None:
+    session.info.pop("govoplan_policy_override_resolution", None)
+
+
+def definition_policy_response_payload(
+    state: DefinitionPolicyState,
+) -> dict[str, Any]:
+    local_policy: Mapping[str, Any] = state.local_policy
+    if state.row is not None:
+        _, malformed = validate_definition_policy(state.row.policy)
+        if malformed:
+            local_policy = {"configuration_status": "invalid_fail_closed"}
+    return {
+        "id": state.row.id if state.row is not None else None,
+        "revision": state.row.revision if state.row is not None else None,
+        "policy": dict(local_policy),
+        "effective_policy": dict(state.effective.limits),
+        "parent_policy": dict(state.parent.limits),
+        "source_path": [step.to_dict() for step in state.effective.source_path],
+        "diagnostics": [dict(item) for item in state.effective.diagnostics],
+    }
+
+
+__all__ = [
+    "DEFINITION_POLICY_FIELDS",
+    "DefinitionPolicyError",
+    "DefinitionPolicyState",
+    "definition_policy_response_payload",
+    "definition_policy_state",
+    "remove_definition_policy",
+    "save_definition_policy",
+]

src/govoplan_policy/backend/manifest.py

@@ -1,6 +1,15 @@
 from __future__ import annotations
 
-from govoplan_core.core.access import CAPABILITY_AUTH_PERMISSION_EVALUATOR, CAPABILITY_AUTH_PRINCIPAL_RESOLVER
+from pathlib import Path
+
+from govoplan_core.core.access import (
+    CAPABILITY_AUTH_PERMISSION_EVALUATOR,
+    CAPABILITY_AUTH_PRINCIPAL_RESOLVER,
+)
+from govoplan_core.core.module_guards import (
+    drop_table_retirement_provider,
+    persistent_table_uninstall_guard,
+)
 from govoplan_core.core.policy import (
     CAPABILITY_POLICY_DEFINITION_GOVERNANCE,
     CAPABILITY_POLICY_PRIVACY_RETENTION,
@@ -8,11 +17,14 @@ from govoplan_core.core.policy import (
 )
 from govoplan_core.core.modules import (
     FrontendModule,
+    MigrationSpec,
     ModuleContext,
     ModuleInterfaceProvider,
     ModuleManifest,
 )
 from govoplan_core.core.views import ViewSurface
+from govoplan_core.db.base import Base
+from govoplan_policy.backend.db import models as policy_models
 
 
 def _route_factory(context: ModuleContext):
@@ -31,7 +43,9 @@ def _privacy_retention_service(context: ModuleContext) -> object:
 
 def _scheduling_participant_privacy_policy(context: ModuleContext) -> object:
     del context
-    from govoplan_policy.backend.scheduling_privacy import SqlSchedulingParticipantPrivacyPolicy
+    from govoplan_policy.backend.scheduling_privacy import (
+        SqlSchedulingParticipantPrivacyPolicy,
+    )
 
     return SqlSchedulingParticipantPrivacyPolicy()
 
@@ -49,7 +63,10 @@ manifest = ModuleManifest(
     id="policy",
     name="Policy",
     version="0.1.9",
-    required_capabilities=(CAPABILITY_AUTH_PRINCIPAL_RESOLVER, CAPABILITY_AUTH_PERMISSION_EVALUATOR),
+    required_capabilities=(
+        CAPABILITY_AUTH_PRINCIPAL_RESOLVER,
+        CAPABILITY_AUTH_PERMISSION_EVALUATOR,
+    ),
     provides_interfaces=(
         ModuleInterfaceProvider(
             name="policy.definition_governance",
@@ -57,14 +74,58 @@ manifest = ModuleManifest(
         ),
     ),
     route_factory=_route_factory,
+    migration_spec=MigrationSpec(
+        module_id="policy",
+        metadata=Base.metadata,
+        script_location=str(Path(__file__).with_name("migrations") / "versions"),
+        retirement_supported=True,
+        retirement_provider=drop_table_retirement_provider(
+            policy_models.PolicyOverride,
+            label="Policy overrides",
+        ),
+        retirement_notes=(
+            "Destructive retirement removes explicit definition and View "
+            "policy overrides after a database snapshot."
+        ),
+    ),
+    uninstall_guard_providers=(
+        persistent_table_uninstall_guard(
+            policy_models.PolicyOverride,
+            label="Policy overrides",
+        ),
+    ),
     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(
-            ViewSurface(id="policy.admin.tenant-retention", module_id="policy", kind="section", label="Tenant retention", order=80),
+                id="policy.admin.system-retention",
-            ViewSurface(id="policy.admin.group-retention", module_id="policy", kind="section", label="Group retention", order=80),
+                module_id="policy",
-            ViewSurface(id="policy.admin.user-retention", module_id="policy", kind="section", label="User retention", order=80),
+                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={

src/govoplan_policy/backend/migrations/__init__.py

@@ -0,0 +1 @@
+"""Policy database migrations."""

src/govoplan_policy/backend/migrations/versions/__init__.py

@@ -0,0 +1 @@
+"""Policy migration revisions."""

src/govoplan_policy/backend/migrations/versions/a9c4e7b2d5f8_v0114_policy_overrides.py

@@ -0,0 +1,75 @@
+"""Add governed hierarchical Policy overrides.
+
+Revision ID: a9c4e7b2d5f8
+Revises: None
+Create Date: 2026-07-31 00:00:00.000000
+"""
+
+from __future__ import annotations
+
+from alembic import op
+import sqlalchemy as sa
+
+
+revision = "a9c4e7b2d5f8"
+down_revision = None
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "policy_overrides",
+        sa.Column("id", sa.String(length=36), nullable=False),
+        sa.Column("policy_family", sa.String(length=40), nullable=False),
+        sa.Column("target_key", sa.String(length=120), nullable=False),
+        sa.Column("tenant_id", sa.String(length=36), nullable=True),
+        sa.Column("scope_type", sa.String(length=20), nullable=False),
+        sa.Column("scope_id", sa.String(length=255), nullable=True),
+        sa.Column("scope_key", sa.String(length=320), nullable=False),
+        sa.Column("policy", sa.JSON(), nullable=False),
+        sa.Column("revision", sa.Integer(), nullable=False),
+        sa.Column("created_by", sa.String(length=255), nullable=True),
+        sa.Column("updated_by", sa.String(length=255), nullable=True),
+        sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False),
+        sa.PrimaryKeyConstraint("id", name=op.f("pk_policy_overrides")),
+        sa.UniqueConstraint(
+            "policy_family",
+            "target_key",
+            "scope_key",
+            name="uq_policy_override_family_target_scope",
+        ),
+    )
+    for column in ("tenant_id", "scope_id", "created_by", "updated_by"):
+        op.create_index(
+            op.f(f"ix_policy_overrides_{column}"),
+            "policy_overrides",
+            [column],
+            unique=False,
+        )
+    op.create_index(
+        "ix_policy_overrides_resolution",
+        "policy_overrides",
+        [
+            "policy_family",
+            "target_key",
+            "tenant_id",
+            "scope_type",
+            "scope_id",
+        ],
+        unique=False,
+    )
+
+
+def downgrade() -> None:
+    op.drop_index(
+        "ix_policy_overrides_resolution",
+        table_name="policy_overrides",
+    )
+    for column in ("updated_by", "created_by", "scope_id", "tenant_id"):
+        op.drop_index(
+            op.f(f"ix_policy_overrides_{column}"),
+            table_name="policy_overrides",
+        )
+    op.drop_table("policy_overrides")

src/govoplan_policy/backend/policy_overrides.py

@@ -0,0 +1,202 @@
+from __future__ import annotations
+
+from collections.abc import Iterable
+from typing import Any
+
+from sqlalchemy import or_
+from sqlalchemy.orm import Session
+
+from govoplan_policy.backend.db.models import PolicyOverride
+
+
+POLICY_SCOPE_TYPES = frozenset({"system", "tenant", "group", "user"})
+POLICY_FAMILIES = frozenset({"definition", "view"})
+
+
+class PolicyOverrideError(ValueError):
+    pass
+
+
+def normalize_policy_target(policy_family: str, target_key: str) -> tuple[str, str]:
+    family = policy_family.strip().casefold()
+    target = target_key.strip().casefold()
+    if family not in POLICY_FAMILIES:
+        raise PolicyOverrideError("Policy family must be definition or view")
+    if not target or len(target) > 120:
+        raise PolicyOverrideError("Policy target must contain 1 to 120 characters")
+    if family == "view" and target != "*":
+        raise PolicyOverrideError("View policy uses the shared '*' target")
+    return family, target
+
+
+def normalize_policy_scope(
+    *,
+    tenant_id: str,
+    scope_type: str,
+    scope_id: str | None,
+) -> tuple[str | None, str | None, str]:
+    clean_scope = scope_type.strip().casefold()
+    clean_id = str(scope_id or "").strip() or None
+    if clean_scope not in POLICY_SCOPE_TYPES:
+        raise PolicyOverrideError("Policy scope must be system, tenant, group, or user")
+    if clean_scope == "system":
+        if clean_id is not None:
+            raise PolicyOverrideError("System policy cannot declare a scope ID")
+        return None, None, "system"
+    if clean_scope == "tenant":
+        if clean_id not in {None, tenant_id}:
+            raise PolicyOverrideError("Tenant policy must target the active tenant")
+        return tenant_id, tenant_id, f"tenant:{tenant_id}"
+    if clean_id is None:
+        raise PolicyOverrideError(
+            f"{clean_scope.capitalize()} policy requires a scope ID"
+        )
+    return tenant_id, clean_id, f"{clean_scope}:{tenant_id}:{clean_id}"
+
+
+def get_policy_override(
+    session: Session,
+    *,
+    policy_family: str,
+    target_key: str,
+    tenant_id: str,
+    scope_type: str,
+    scope_id: str | None = None,
+) -> PolicyOverride | None:
+    family, target = normalize_policy_target(policy_family, target_key)
+    _, _, scope_key = normalize_policy_scope(
+        tenant_id=tenant_id,
+        scope_type=scope_type,
+        scope_id=scope_id,
+    )
+    return (
+        session.query(PolicyOverride)
+        .filter(
+            PolicyOverride.policy_family == family,
+            PolicyOverride.target_key == target,
+            PolicyOverride.scope_key == scope_key,
+        )
+        .first()
+    )
+
+
+def set_policy_override(
+    session: Session,
+    *,
+    policy_family: str,
+    target_key: str,
+    tenant_id: str,
+    scope_type: str,
+    scope_id: str | None,
+    policy: Any,
+    actor_id: str | None,
+) -> PolicyOverride:
+    family, target = normalize_policy_target(policy_family, target_key)
+    row_tenant_id, clean_scope_id, scope_key = normalize_policy_scope(
+        tenant_id=tenant_id,
+        scope_type=scope_type,
+        scope_id=scope_id,
+    )
+    row = (
+        session.query(PolicyOverride)
+        .filter(
+            PolicyOverride.policy_family == family,
+            PolicyOverride.target_key == target,
+            PolicyOverride.scope_key == scope_key,
+        )
+        .first()
+    )
+    if row is None:
+        row = PolicyOverride(
+            policy_family=family,
+            target_key=target,
+            tenant_id=row_tenant_id,
+            scope_type=scope_type.strip().casefold(),
+            scope_id=clean_scope_id,
+            scope_key=scope_key,
+            policy=policy,
+            created_by=actor_id,
+            updated_by=actor_id,
+        )
+        session.add(row)
+    else:
+        row.policy = policy
+        row.revision += 1
+        row.updated_by = actor_id
+    session.flush()
+    return row
+
+
+def delete_policy_override(session: Session, row: PolicyOverride) -> None:
+    session.delete(row)
+    session.flush()
+
+
+def resolution_policy_overrides(
+    session: Session,
+    *,
+    policy_family: str,
+    target_keys: Iterable[str],
+    tenant_id: str,
+    group_ids: Iterable[str] = (),
+    user_ids: Iterable[str] = (),
+) -> tuple[PolicyOverride, ...]:
+    family = policy_family.strip().casefold()
+    targets = tuple(
+        dict.fromkeys(
+            normalize_policy_target(family, target)[1] for target in target_keys
+        )
+    )
+    groups = tuple(sorted({str(value) for value in group_ids if str(value)}))
+    users = tuple(sorted({str(value) for value in user_ids if str(value)}))
+    scope_filters = [
+        PolicyOverride.scope_key == "system",
+        PolicyOverride.scope_key == f"tenant:{tenant_id}",
+    ]
+    if groups:
+        scope_filters.append(
+            PolicyOverride.scope_key.in_(
+                [f"group:{tenant_id}:{group_id}" for group_id in groups]
+            )
+        )
+    if users:
+        scope_filters.append(
+            PolicyOverride.scope_key.in_(
+                [f"user:{tenant_id}:{user_id}" for user_id in users]
+            )
+        )
+    rows = (
+        session.query(PolicyOverride)
+        .filter(
+            PolicyOverride.policy_family == family,
+            PolicyOverride.target_key.in_(targets),
+            or_(*scope_filters),
+        )
+        .all()
+    )
+    target_order = {target: index for index, target in enumerate(targets)}
+    scope_order = {"system": 0, "tenant": 1, "group": 2, "user": 3}
+    return tuple(
+        sorted(
+            rows,
+            key=lambda row: (
+                scope_order.get(row.scope_type, 99),
+                row.scope_id or "",
+                target_order.get(row.target_key, 99),
+                row.id,
+            ),
+        )
+    )
+
+
+__all__ = [
+    "POLICY_FAMILIES",
+    "POLICY_SCOPE_TYPES",
+    "PolicyOverrideError",
+    "delete_policy_override",
+    "get_policy_override",
+    "normalize_policy_scope",
+    "normalize_policy_target",
+    "resolution_policy_overrides",
+    "set_policy_override",
+]

tests/test_definition_policy_overrides.py

@@ -0,0 +1,215 @@
+from __future__ import annotations
+
+import unittest
+
+from sqlalchemy import create_engine
+from sqlalchemy.orm import Session, sessionmaker
+
+from govoplan_core.core.access import PrincipalRef
+from govoplan_core.core.policy import DefinitionGovernanceRequest, DefinitionScopeRef
+from govoplan_policy.backend.db.models import PolicyOverride
+from govoplan_policy.backend.definition_governance import (
+    DEFINITION_POLICY_FIELDS,
+    DefinitionGovernancePolicyProvider,
+)
+from govoplan_policy.backend.definition_policy_service import (
+    DefinitionPolicyError,
+    definition_policy_response_payload,
+    definition_policy_state,
+    save_definition_policy,
+)
+
+
+class DefinitionPolicyOverrideTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.engine = create_engine("sqlite+pysqlite:///:memory:")
+        PolicyOverride.__table__.create(self.engine)
+        self.session_factory = sessionmaker(
+            bind=self.engine,
+            expire_on_commit=False,
+        )
+        self.session: Session = self.session_factory()
+        self.provider = DefinitionGovernancePolicyProvider()
+        self.actor = PrincipalRef(
+            account_id="account-1",
+            membership_id="user-1",
+            tenant_id="tenant-1",
+            group_ids=frozenset({"group-1"}),
+        )
+
+    def tearDown(self) -> None:
+        self.session.close()
+        self.engine.dispose()
+
+    def _save(
+        self,
+        *,
+        scope_type: str,
+        policy: object,
+        scope_id: str | None = None,
+        module_id: str = "dataflow",
+    ):
+        return save_definition_policy(
+            self.session,
+            module_id=module_id,
+            tenant_id="tenant-1",
+            scope_type=scope_type,
+            scope_id=scope_id,
+            policy=policy,
+            actor_id="account-1",
+        )
+
+    def _resolve(self, *, action: str, definition_scope: DefinitionScopeRef):
+        return self.provider.resolve_definition_action(
+            self.session,
+            request=DefinitionGovernanceRequest(
+                module_id="dataflow",
+                definition_ref="pipeline:1",
+                tenant_id="tenant-1",
+                definition_scope=definition_scope,
+                target_scope=DefinitionScopeRef("tenant", "tenant-1"),
+                definition_kind="flow",
+                action=action,  # type: ignore[arg-type]
+                actor=self.actor,
+                status="active",
+                inherit_to_lower_scopes=True,
+                allow_run=True,
+                allow_reuse=True,
+                allow_automation=True,
+            ),
+        )
+
+    def test_persists_and_revises_an_explicit_policy(self) -> None:
+        first = self._save(scope_type="tenant", policy={"allow_run": False})
+        second = self._save(
+            scope_type="tenant",
+            policy={"allow_run": False, "allow_reuse": False},
+        )
+
+        self.assertEqual(first.row.id, second.row.id)
+        self.assertEqual(2, second.row.revision)
+        self.assertEqual(
+            {"allow_run": False, "allow_reuse": False},
+            second.local_policy,
+        )
+
+    def test_system_and_tenant_limits_apply_to_group_and_user_targets(self) -> None:
+        self._save(
+            scope_type="system",
+            module_id="*",
+            policy={"allow_automation": False},
+        )
+        self._save(
+            scope_type="tenant",
+            policy={"allow_reuse": False},
+        )
+
+        group = definition_policy_state(
+            self.session,
+            module_id="dataflow",
+            tenant_id="tenant-1",
+            scope_type="group",
+            scope_id="group-1",
+        )
+        user = definition_policy_state(
+            self.session,
+            module_id="dataflow",
+            tenant_id="tenant-1",
+            scope_type="user",
+            scope_id="user-1",
+        )
+
+        for state in (group, user):
+            self.assertFalse(state.effective.limits["allow_automation"])
+            self.assertFalse(state.effective.limits["allow_reuse"])
+            self.assertEqual(
+                ["system", "tenant:tenant-1"],
+                [step.path for step in state.effective.source_path],
+            )
+
+    def test_lower_scopes_cannot_broaden_parent_restrictions(self) -> None:
+        self._save(scope_type="tenant", policy={"allow_run": False})
+
+        with self.assertRaisesRegex(
+            DefinitionPolicyError,
+            "cannot broaden parent restrictions: allow_run",
+        ):
+            self._save(
+                scope_type="group",
+                scope_id="group-1",
+                policy={"allow_run": True},
+            )
+
+    def test_explicit_policy_restricts_provider_actions(self) -> None:
+        self._save(
+            scope_type="tenant",
+            policy={"allow_edit": False, "allow_reuse": False},
+        )
+
+        edit = self._resolve(
+            action="edit",
+            definition_scope=DefinitionScopeRef("tenant", "tenant-1"),
+        )
+        reuse = self._resolve(
+            action="reuse",
+            definition_scope=DefinitionScopeRef("group", "group-1"),
+        )
+
+        self.assertFalse(edit.allowed)
+        self.assertFalse(reuse.allowed)
+        self.assertEqual(
+            "Editing is disabled by explicit Policy restrictions.",
+            edit.reason,
+        )
+        self.assertEqual("tenant:tenant-1", edit.source_path[0].path)
+
+    def test_malformed_persisted_policy_fails_closed_and_is_redacted(self) -> None:
+        row = PolicyOverride(
+            policy_family="definition",
+            target_key="dataflow",
+            tenant_id="tenant-1",
+            scope_type="tenant",
+            scope_id="tenant-1",
+            scope_key="tenant:tenant-1",
+            policy={"allow_run": "definitely", "secret": "do-not-echo"},
+        )
+        self.session.add(row)
+        self.session.flush()
+
+        decision = self._resolve(
+            action="run",
+            definition_scope=DefinitionScopeRef("tenant", "tenant-1"),
+        )
+        state = definition_policy_state(
+            self.session,
+            module_id="dataflow",
+            tenant_id="tenant-1",
+            scope_type="tenant",
+        )
+        payload = definition_policy_response_payload(state)
+
+        self.assertFalse(decision.allowed)
+        self.assertEqual(
+            {field: False for field in DEFINITION_POLICY_FIELDS},
+            decision.details["effective_limits"],
+        )
+        self.assertEqual(
+            [
+                {
+                    "code": "definition_policy.invalid",
+                    "scope": "tenant:tenant-1",
+                    "target_key": "dataflow",
+                }
+            ],
+            decision.details["policy_diagnostics"],
+        )
+        self.assertEqual(
+            {"configuration_status": "invalid_fail_closed"},
+            payload["policy"],
+        )
+        self.assertNotIn("secret", repr(decision.to_dict()))
+        self.assertNotIn("do-not-echo", repr(payload))
+
+
+if __name__ == "__main__":
+    unittest.main()