feat(policy): govern delegation and review escalation

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,