feat(policy): add delegation and escalation contracts
Module Package Release / publish-packages (push) Failing after 6s

This commit is contained in:
2026-08-22 03:12:30 +02:00
parent a9035c4c3b
commit 0c1358b862
6 changed files with 83 additions and 4 deletions
+32
View File
@@ -41,10 +41,12 @@ ViewGovernanceAction = Literal[
"workflow_activate",
]
FunctionAssignmentChangeKind = Literal["request", "grant"]
FunctionAssignmentReviewStep = Literal["holder", "authority", "recipient"]
FunctionAssignmentGovernanceAction = Literal[
"submit",
"approve_holder",
"approve_authority",
"approve_escalation",
"accept_recipient",
"request_changes",
"respond",
@@ -419,6 +421,20 @@ class FunctionAssignmentGovernanceRequest:
context: Mapping[str, Any] = field(default_factory=dict)
@dataclass(frozen=True, slots=True)
class FunctionAssignmentEscalationRule:
step: FunctionAssignmentReviewStep
target_function_id: str
timeout_hours: int
def to_dict(self) -> dict[str, Any]:
return {
"step": self.step,
"target_function_id": self.target_function_id,
"timeout_hours": self.timeout_hours,
}
@dataclass(frozen=True, slots=True)
class FunctionAssignmentGovernanceDecision:
allowed: bool
@@ -431,6 +447,10 @@ class FunctionAssignmentGovernanceDecision:
separation_of_duties: bool = True
quorum: int = 1
maximum_validity_days: int | None = None
delegation_allowed: bool = False
maximum_delegation_depth: int = 0
maximum_delegated_validity_days: int | None = None
escalation_rules: tuple[FunctionAssignmentEscalationRule, ...] = ()
request_expiry_hours: int = 336
source_path: tuple[PolicySourceStep, ...] = ()
requirements: tuple[str, ...] = ()
@@ -448,12 +468,24 @@ class FunctionAssignmentGovernanceDecision:
"separation_of_duties": self.separation_of_duties,
"quorum": self.quorum,
"maximum_validity_days": self.maximum_validity_days,
"delegation_allowed": self.delegation_allowed,
"maximum_delegation_depth": self.maximum_delegation_depth,
"maximum_delegated_validity_days": (
self.maximum_delegated_validity_days
),
"escalation_rules": [rule.to_dict() for rule in self.escalation_rules],
"request_expiry_hours": self.request_expiry_hours,
"source_path": [step.to_dict() for step in self.source_path],
"requirements": list(self.requirements),
"details": dict(self.details),
}
def escalation_rule(
self,
step: FunctionAssignmentReviewStep,
) -> FunctionAssignmentEscalationRule | None:
return next((rule for rule in self.escalation_rules if rule.step == step), None)
@runtime_checkable
class FunctionAssignmentGovernancePolicy(Protocol):