From 5753488375f419753bba2df13a45c92ab93a9ed7 Mon Sep 17 00:00:00 2001 From: Albrecht Degering Date: Sat, 22 Aug 2026 03:12:30 +0200 Subject: [PATCH] feat(policy): govern delegation and review escalation --- docs/POLICY_DECISION_PROVENANCE.md | 15 +++ pyproject.toml | 4 +- .../backend/function_assignment_governance.py | 126 +++++++++++++++++- src/govoplan_policy/backend/manifest.py | 26 +++- tests/test_function_assignment_governance.py | 52 ++++++++ webui/package.json | 2 +- 6 files changed, 217 insertions(+), 8 deletions(-) diff --git a/docs/POLICY_DECISION_PROVENANCE.md b/docs/POLICY_DECISION_PROVENANCE.md index ce834ae..a55888f 100644 --- a/docs/POLICY_DECISION_PROVENANCE.md +++ b/docs/POLICY_DECISION_PROVENANCE.md @@ -130,3 +130,18 @@ bounded outcome counts in audit evidence. The shared core WebUI helper `PolicySourcePath` renders the source path shape for module UIs. Modules may use their own field layout, but the data contract should remain this shape. +# Function assignment delegation and escalation + +The `policy.functionAssignmentGovernance` decision includes the effective +`delegation_allowed`, `maximum_delegation_depth`, and +`maximum_delegated_validity_days` values plus zero or more per-step escalation +rules. Each rule binds `holder`, `authority`, or `recipient` review to one exact +target function and a bounded timeout. Consumers must treat the decision as a +current limit, not a captured grant: IDM rechecks it across the complete source +chain at every consequential transition. + +An elapsed timeout does not change the approval result. IDM records an explicit +escalated state and the target function; Policy authorizes only a current holder +of that target for the escalated decision. Missing, malformed, vacant, expired, +cyclic, over-depth, or tightened routes fail closed with their reason preserved +in the decision and transition evidence. diff --git a/pyproject.toml b/pyproject.toml index 5582cf5..589406d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,13 +4,13 @@ build-backend = "setuptools.build_meta" [project] name = "govoplan-policy" -version = "0.1.19" +version = "0.1.20" description = "GovOPlaN policy platform module." readme = "README.md" requires-python = ">=3.12" authors = [{ name = "GovOPlaN" }] dependencies = [ - "govoplan-core>=0.1.20", + "govoplan-core>=0.1.29", ] [tool.setuptools.packages.find] diff --git a/src/govoplan_policy/backend/function_assignment_governance.py b/src/govoplan_policy/backend/function_assignment_governance.py index fed2096..9690148 100644 --- a/src/govoplan_policy/backend/function_assignment_governance.py +++ b/src/govoplan_policy/backend/function_assignment_governance.py @@ -3,6 +3,7 @@ from __future__ import annotations from collections.abc import Mapping from govoplan_core.core.policy import ( + FunctionAssignmentEscalationRule, FunctionAssignmentGovernanceDecision, FunctionAssignmentGovernanceRequest, PolicySourceStep, @@ -50,6 +51,25 @@ class FunctionAssignmentGovernancePolicyProvider: ), ) authority_function_id = _text(policy.get("authority_function_id")) + delegation_allowed = _bool( + policy.get("delegation_allowed"), + default=False, + ) + maximum_delegation_depth = ( + _bounded_int( + policy.get("maximum_delegation_depth"), + default=1, + minimum=1, + maximum=20, + ) + if delegation_allowed + else 0 + ) + maximum_delegated_validity_days = _optional_positive_int( + policy.get("maximum_delegated_validity_days"), + maximum=3650, + ) + escalation_rules, escalation_requirements = _escalation_rules(policy) requirements: list[str] = [] if "authority" in required_steps and authority_function_id is None: requirements.append("authority_function") @@ -59,6 +79,7 @@ class FunctionAssignmentGovernancePolicyProvider: ) if evidence_required and not request.context.get("has_evidence"): requirements.append("evidence") + requirements.extend(escalation_requirements) allowed, reason = _action_decision( request, profile=profile, @@ -79,6 +100,10 @@ class FunctionAssignmentGovernancePolicyProvider: required_steps=required_steps, authority_function_id=authority_function_id, evidence_required=evidence_required, + delegation_allowed=delegation_allowed, + maximum_delegation_depth=maximum_delegation_depth, + maximum_delegated_validity_days=maximum_delegated_validity_days, + escalation_rules=escalation_rules, requirements=tuple(requirements), ) @@ -106,19 +131,44 @@ def _action_decision( return True, None if profile == "authority_only": allowed = bool(context.get("actor_is_authority")) - reason = "Only the designated authority may initiate this grant." + reason = _route_reason( + context, + "authority", + "Only the designated authority may initiate this grant.", + ) else: allowed = bool(context.get("actor_is_holder")) - reason = "An effective function holder must initiate this grant." + reason = _route_reason( + context, + "holder", + "An effective function holder must initiate this grant.", + ) return allowed, None if allowed else reason if action == "approve_holder": allowed = "holder" in required_steps and bool(context.get("actor_is_holder")) - return allowed, None if allowed else "A current holder must approve." + return allowed, None if allowed else _route_reason( + context, + "holder", + "A current holder must approve.", + ) if action == "approve_authority": allowed = "authority" in required_steps and bool( context.get("actor_is_authority") ) - return allowed, None if allowed else "The designated authority must approve." + return allowed, None if allowed else _route_reason( + context, + "authority", + "The designated authority must approve.", + ) + if action == "approve_escalation": + allowed = request.current_state == "escalated" and bool( + context.get("actor_is_escalation_target") + ) + return allowed, None if allowed else _route_reason( + context, + "escalation", + "A current holder of the explicit escalation target must approve.", + ) if action == "accept_recipient": allowed = "recipient" in required_steps and bool( context.get("candidate_is_actor") @@ -134,6 +184,13 @@ def _action_decision( elif request.current_state == "awaiting_recipient": allowed = bool(context.get("candidate_is_actor")) reason = "Only the candidate may act at recipient acceptance." + elif request.current_state == "escalated": + allowed = bool(context.get("actor_is_escalation_target")) + reason = _route_reason( + context, + "escalation", + "Only a current holder of the explicit escalation target may act.", + ) else: allowed = False reason = "The current state does not accept this review action." @@ -196,6 +253,10 @@ def _decision( required_steps: tuple[str, ...] = (), authority_function_id: str | None = None, evidence_required: bool = False, + delegation_allowed: bool = False, + maximum_delegation_depth: int = 0, + maximum_delegated_validity_days: int | None = None, + escalation_rules: tuple[FunctionAssignmentEscalationRule, ...] = (), requirements: tuple[str, ...] = (), ) -> FunctionAssignmentGovernanceDecision: recipient_required = "recipient" in required_steps @@ -216,6 +277,10 @@ def _decision( policy.get("maximum_validity_days"), maximum=3650, ), + delegation_allowed=delegation_allowed, + maximum_delegation_depth=maximum_delegation_depth, + maximum_delegated_validity_days=maximum_delegated_validity_days, + escalation_rules=escalation_rules, request_expiry_hours=_bounded_int( policy.get("request_expiry_hours"), default=336, @@ -236,6 +301,7 @@ def _decision( "function_id": request.function_id, "kind": request.kind, "action": request.action, + "actor_routes": dict(request.context.get("actor_routes") or {}), }, ) @@ -250,6 +316,9 @@ def _requirements_reason(requirements: list[str]) -> str: "authority_function": "a designated authority function", "evidence": "the required evidence", "valid_profile": "a supported governance profile", + "escalation_holder": "a valid holder-step escalation rule", + "escalation_authority": "a valid authority-step escalation rule", + "escalation_recipient": "a valid recipient-step escalation rule", } return ( "Submission requires " @@ -258,6 +327,55 @@ def _requirements_reason(requirements: list[str]) -> str: ) +def _escalation_rules( + policy: Mapping[str, object], +) -> tuple[tuple[FunctionAssignmentEscalationRule, ...], list[str]]: + raw = policy.get("escalation") + if raw is None: + return (), [] + if not isinstance(raw, Mapping): + return (), ["escalation_holder"] + rules: list[FunctionAssignmentEscalationRule] = [] + requirements: list[str] = [] + for step in ("holder", "authority", "recipient"): + value = raw.get(step) + if value is None: + continue + if not isinstance(value, Mapping): + requirements.append(f"escalation_{step}") + continue + target_function_id = _text(value.get("target_function_id")) + timeout_hours = _optional_positive_int( + value.get("timeout_hours"), + maximum=8760, + ) + if target_function_id is None or timeout_hours is None: + requirements.append(f"escalation_{step}") + continue + rules.append( + FunctionAssignmentEscalationRule( + step=step, # type: ignore[arg-type] + target_function_id=target_function_id, + timeout_hours=timeout_hours, + ) + ) + return tuple(rules), requirements + + +def _route_reason( + context: Mapping[str, object], + route: str, + fallback: str, +) -> str: + routes = context.get("actor_routes") + if not isinstance(routes, Mapping): + return fallback + value = routes.get(route) + if not isinstance(value, Mapping): + return fallback + return _text(value.get("reason")) or fallback + + def _has_scope( request: FunctionAssignmentGovernanceRequest, scope: str, diff --git a/src/govoplan_policy/backend/manifest.py b/src/govoplan_policy/backend/manifest.py index fc3ee9b..8c3db68 100644 --- a/src/govoplan_policy/backend/manifest.py +++ b/src/govoplan_policy/backend/manifest.py @@ -150,7 +150,7 @@ POLICY_IMPACT_DETAILS_SCOPE = "policy:impact:details" manifest = ModuleManifest( id="policy", name="Policy", - version="0.1.19", + version="0.1.20", permissions=( PermissionDefinition( scope=ACCESS_EXPLANATION_SUBJECT_SCOPE, @@ -307,6 +307,30 @@ manifest = ModuleManifest( ], }, ), + DocumentationTopic( + id="policy.function-assignment-delegation-escalation", + title="Govern function delegation and review escalation", + summary="Policy bounds complete delegation chains and defines explicit target functions for overdue assignment reviews.", + body=( + "Tenant defaults and function settings may allow delegation, cap its chain depth and validity, and configure a holder, authority, or recipient review timeout with an exact escalation target function. IDM rechecks the complete current chain and the effective Policy at submission, every decision, recovery, and application. A tightened limit invalidates an old route with an explanation. A timeout creates a visible escalated state but never substitutes an approver or completes the review; a current target-function holder must decide explicitly. Malformed or incomplete escalation rules fail closed." + ), + documentation_types=("admin", "user"), + audience=("tenant_admin", "policy_admin", "access_admin", "user"), + related_modules=("idm", "organizations", "workflow_engine", "notifications", "audit"), + metadata={ + "kind": "reference", + "help_contexts": [ + "idm.field.delegation-ceilings", + "idm.field.escalation", + ], + "fields": [ + {"key": "delegation_allowed", "consequence": "Allows governed derived assignments only when Organizations also marks the function delegable."}, + {"key": "maximum_delegation_depth", "consequence": "Rejects longer current chains, including chains accepted before a tighter limit."}, + {"key": "maximum_delegated_validity_days", "consequence": "Caps each delegated validity window in addition to its source window."}, + {"key": "escalation.", "consequence": "Pins a target function and deadline without granting or substituting approval."}, + ], + }, + ), DocumentationTopic( id="policy.effective-decisions-and-provenance", title="Understand effective policy decisions", diff --git a/tests/test_function_assignment_governance.py b/tests/test_function_assignment_governance.py index 5cbe58b..32750c3 100644 --- a/tests/test_function_assignment_governance.py +++ b/tests/test_function_assignment_governance.py @@ -132,6 +132,58 @@ class FunctionAssignmentGovernancePolicyTests(unittest.TestCase): self.assertTrue(responder.allowed) self.assertFalse(unrelated.allowed) + def test_delegation_ceilings_and_escalation_rules_are_bounded(self) -> None: + decision = self.resolve( + function_settings={ + "assignment_governance": { + "request_profile": "holder_with_authority_clearance", + "authority_function_id": "authority-1", + "delegation_allowed": True, + "maximum_delegation_depth": 3, + "maximum_delegated_validity_days": 45, + "escalation": { + "holder": { + "target_function_id": "escalation-1", + "timeout_hours": 24, + } + }, + } + } + ) + + self.assertTrue(decision.delegation_allowed) + self.assertEqual(3, decision.maximum_delegation_depth) + self.assertEqual(45, decision.maximum_delegated_validity_days) + self.assertEqual("escalation-1", decision.escalation_rules[0].target_function_id) + self.assertEqual(24, decision.escalation_rules[0].timeout_hours) + + def test_escalated_review_requires_explicit_target_holder(self) -> None: + allowed = self.resolve( + action="approve_escalation", + current_state="escalated", + context={ + "actor_is_escalation_target": True, + "actor_routes": {"escalation": {"effective": True}}, + }, + ) + unavailable = self.resolve( + action="approve_escalation", + current_state="escalated", + context={ + "actor_is_escalation_target": False, + "actor_routes": { + "escalation": { + "effective": False, + "reason": "The target function is vacant.", + } + }, + }, + ) + + self.assertTrue(allowed.allowed) + self.assertFalse(unavailable.allowed) + self.assertEqual("The target function is vacant.", unavailable.reason) + if __name__ == "__main__": unittest.main() diff --git a/webui/package.json b/webui/package.json index 8259faa..4f84766 100644 --- a/webui/package.json +++ b/webui/package.json @@ -1,6 +1,6 @@ { "name": "@govoplan/policy-webui", - "version": "0.1.19", + "version": "0.1.20", "private": true, "type": "module", "main": "src/index.ts",