Update infrastructure/policies.py

infrastructure/policies.py

@@ -1,14 +1,5 @@
-# agentic_reliability_framework/infrastructure/policies.py
 """
 Policy Algebra – Composable, typed policies for infrastructure governance.
-
-This module defines a composable policy system using a monoid-like structure.
-Policies can be combined (AND, OR) and evaluated against intents. The algebra
-enables building complex rules from simple primitives while maintaining
-deterministic evaluation.
-
-The design draws from knowledge engineering (rule-based systems), decision
-engineering (explicit trade-offs), and platform engineering (pluggable backends).
 """
 
 from __future__ import annotations
@@ -19,7 +10,8 @@ from typing import Any, Callable, Dict, List, Optional, Protocol, Set, TypeVar,
 
 from pydantic import BaseModel
 
-from agentic_reliability_framework.infrastructure.intents import (
+# Relative imports for intents
+from .intents import (
     InfrastructureIntent,
     ProvisionResourceIntent,
     GrantAccessIntent,
@@ -28,48 +20,28 @@ from agentic_reliability_framework.infrastructure.intents import (
     Environment,
 )
 
-# -----------------------------------------------------------------------------
-# Type Aliases for Readability
-# -----------------------------------------------------------------------------
 Violation = str
 EvalResult = List[Violation]
 
-# -----------------------------------------------------------------------------
-# Abstract Policy Class – using Composite pattern
-# -----------------------------------------------------------------------------
 class Policy(ABC):
-    """Abstract base for all policies. Evaluates an intent and returns violations."""
-
     @abstractmethod
     def evaluate(self, intent: InfrastructureIntent) -> EvalResult:
-        """Return list of violations. Empty list means the policy is satisfied."""
         pass
 
-    # -------------------------------------------------------------------------
-    # Combinators (returning new composite policies)
-    # -------------------------------------------------------------------------
     def __and__(self, other: Policy) -> Policy:
-        """Logical AND: violations from both policies."""
         return AndPolicy(self, other)
 
     def __or__(self, other: Policy) -> Policy:
-        """Logical OR: violations only if both policies produce violations."""
         return OrPolicy(self, other)
 
     def __invert__(self) -> Policy:
-        """Logical NOT: violations if the original policy yields no violations."""
         return NotPolicy(self)
 
-# -----------------------------------------------------------------------------
-# Atomic Policies (Primitives)
-# -----------------------------------------------------------------------------
 class AtomicPolicy(Policy, ABC):
-    """Base class for policies that don't contain other policies."""
     pass
 
 @dataclass(frozen=True)
 class RegionAllowedPolicy(AtomicPolicy):
-    """Ensure the intent's region is in an allowed set."""
     allowed_regions: Set[str]
 
     def evaluate(self, intent: InfrastructureIntent) -> EvalResult:
@@ -80,7 +52,6 @@ class RegionAllowedPolicy(AtomicPolicy):
 
 @dataclass(frozen=True)
 class ResourceTypeRestrictedPolicy(AtomicPolicy):
-    """Forbid certain resource types."""
     forbidden_types: Set[ResourceType]
 
     def evaluate(self, intent: InfrastructureIntent) -> EvalResult:
@@ -91,10 +62,8 @@ class ResourceTypeRestrictedPolicy(AtomicPolicy):
 
 @dataclass(frozen=True)
 class MaxPermissionLevelPolicy(AtomicPolicy):
-    """Limit the maximum permission level that can be granted."""
     max_level: PermissionLevel
 
-    # Permission level ordering (read < write < admin)
     _LEVEL_ORDER = {
         PermissionLevel.READ: 1,
         PermissionLevel.WRITE: 2,
@@ -109,26 +78,13 @@ class MaxPermissionLevelPolicy(AtomicPolicy):
 
 @dataclass(frozen=True)
 class CostThresholdPolicy(AtomicPolicy):
-    """
-    Enforce a maximum monthly cost.
-    Note: This policy requires the cost estimate to be provided externally.
-    We evaluate it only when a cost is supplied via context.
-    """
     max_cost_usd: float
 
     def evaluate(self, intent: InfrastructureIntent, cost: Optional[float] = None) -> EvalResult:
-        # This is a special case – we need cost from the simulator.
-        # We'll handle it by allowing context injection. For composition, we'll use a wrapper.
-        # In practice, we'll evaluate this in the simulator and add violations manually.
-        # For now, we keep it as a marker.
         return []
 
-# -----------------------------------------------------------------------------
-# Composite Policies (using Combinators)
-# -----------------------------------------------------------------------------
 @dataclass(frozen=True)
 class AndPolicy(Policy):
-    """Logical AND of two policies."""
     left: Policy
     right: Policy
 
@@ -137,63 +93,42 @@ class AndPolicy(Policy):
 
 @dataclass(frozen=True)
 class OrPolicy(Policy):
-    """Logical OR – violations only if both sub-policies have violations."""
     left: Policy
     right: Policy
 
     def evaluate(self, intent: InfrastructureIntent) -> EvalResult:
         left_violations = self.left.evaluate(intent)
         right_violations = self.right.evaluate(intent)
-        # If either has no violations, overall no violations (OR satisfied)
         if not left_violations or not right_violations:
             return []
-        # Both have violations – combine them
         return left_violations + right_violations
 
 @dataclass(frozen=True)
 class NotPolicy(Policy):
-    """Logical NOT – violations if inner policy has no violations."""
     inner: Policy
 
     def evaluate(self, intent: InfrastructureIntent) -> EvalResult:
         inner_violations = self.inner.evaluate(intent)
         if inner_violations:
-            return []  # inner failed (had violations), so NOT is satisfied (no violations)
+            return []
         else:
             return ["Condition not met (NOT policy)"]
 
-# -----------------------------------------------------------------------------
-# Context-Aware Evaluation
-# -----------------------------------------------------------------------------
 class PolicyEvaluator:
-    """
-    Evaluates a policy tree with additional context (e.g., cost estimates).
-    This allows policies that depend on external data to be evaluated.
-    """
-
     def __init__(self, root_policy: Policy):
         self._root = root_policy
 
     def evaluate(self, intent: InfrastructureIntent, context: Optional[Dict[str, Any]] = None) -> EvalResult:
-        """
-        Evaluate the policy tree against the intent, using context for dynamic checks.
-        Context may contain 'cost_estimate' for CostThresholdPolicy, etc.
-        """
-        # For simplicity, we traverse the tree and apply context where needed.
-        # A more sophisticated implementation would use a visitor pattern.
         return self._evaluate_recursive(self._root, intent, context or {})
 
     def _evaluate_recursive(self, policy: Policy, intent: InfrastructureIntent, context: Dict[str, Any]) -> EvalResult:
-        # If policy is CostThresholdPolicy, we use context to evaluate.
         if isinstance(policy, CostThresholdPolicy):
             cost = context.get('cost_estimate')
             if cost is not None and cost > policy.max_cost_usd:
                 return [f"Cost ${cost:.2f} exceeds threshold ${policy.max_cost_usd:.2f}"]
             return []
-        # For atomic policies, just evaluate (they don't need context)
         if isinstance(policy, AtomicPolicy):
             return policy.evaluate(intent)
-        # For composites, recurse
         if isinstance(policy, AndPolicy):
             return self._evaluate_recursive(policy.left, intent, context) + self._evaluate_recursive(policy.right, intent, context)
         if isinstance(policy, OrPolicy):
@@ -207,28 +142,20 @@ class PolicyEvaluator:
             if inner:
                 return []
             return ["Condition not met (NOT policy)"]
-        # Fallback
         return []
 
-# -----------------------------------------------------------------------------
-# Policy Builder (DSL) – convenience functions
-# -----------------------------------------------------------------------------
 def allow_all() -> Policy:
-    """Policy that never produces violations."""
     class _AllowAll(AtomicPolicy):
         def evaluate(self, intent: InfrastructureIntent) -> EvalResult:
             return []
     return _AllowAll()
 
 def deny_all() -> Policy:
-    """Policy that always produces a violation."""
     class _DenyAll(AtomicPolicy):
         def evaluate(self, intent: InfrastructureIntent) -> EvalResult:
             return ["Action denied by default policy"]
     return _DenyAll()
 
-# Example: policy = (region_allowed({"eastus"}) & ~resource_type_restricted({ResourceType.KUBERNETES_CLUSTER})) | cost_threshold(500)
-
 __all__ = [
     "Policy",
     "RegionAllowedPolicy",