Create risk_engine.py

infrastructure/risk_engine.py

@@ -0,0 +1,153 @@
+# agentic_reliability_framework/infrastructure/risk_engine.py
+"""
+Risk Scoring Engine – Multi-factor probabilistic risk model.
+
+This module computes a risk score (0-1) for an infrastructure intent by combining
+multiple factors with configurable weights. The model is inspired by Bayesian
+decision theory and multi-criteria decision analysis (MCDA). It produces not only
+a score but also a detailed explanation of each factor's contribution, supporting
+transparency and psychological trust.
+
+The risk engine is designed to be extended with additional factors (e.g., historical
+data, anomaly scores) without changing the core API.
+"""
+
+from typing import Dict, List, Optional, Tuple, Any, Callable
+from dataclasses import dataclass, field
+
+from agentic_reliability_framework.infrastructure.intents import (
+    InfrastructureIntent,
+    ProvisionResourceIntent,
+    GrantAccessIntent,
+    DeployConfigurationIntent,
+    PermissionLevel,
+    Environment,
+)
+
+# -----------------------------------------------------------------------------
+# Factor Definition
+# -----------------------------------------------------------------------------
+@dataclass(frozen=True)
+class RiskFactor:
+    """A single factor contributing to risk, with a weight and a scoring function."""
+    name: str
+    weight: float
+    score_fn: Callable[[InfrastructureIntent, Optional[float], List[str]], float]
+    description: str = ""
+
+    def __call__(self, intent: InfrastructureIntent, cost: Optional[float], violations: List[str]) -> float:
+        return self.score_fn(intent, cost, violations)
+
+# -----------------------------------------------------------------------------
+# Built-in Factors
+# -----------------------------------------------------------------------------
+def intent_type_factor(intent: InfrastructureIntent, cost: Optional[float], violations: List[str]) -> float:
+    """Base risk from intent type."""
+    mapping = {
+        "provision_resource": 0.1,
+        "grant_access": 0.3,
+        "deploy_config": 0.2,
+    }
+    return mapping.get(intent.intent_type, 0.1)
+
+def cost_factor(intent: InfrastructureIntent, cost: Optional[float], violations: List[str]) -> float:
+    """Risk contribution from estimated cost (normalized to [0,1])."""
+    if not isinstance(intent, ProvisionResourceIntent) or cost is None:
+        return 0.0
+    # Normalize: $0 → 0, $5000 → 1 (linear)
+    return min(cost / 5000.0, 1.0)
+
+def permission_factor(intent: InfrastructureIntent, cost: Optional[float], violations: List[str]) -> float:
+    """Risk from permission level being granted."""
+    if not isinstance(intent, GrantAccessIntent):
+        return 0.0
+    mapping = {
+        PermissionLevel.READ: 0.1,
+        PermissionLevel.WRITE: 0.4,
+        PermissionLevel.ADMIN: 0.8,
+    }
+    return mapping.get(intent.permission_level, 0.5)
+
+def scope_factor(intent: InfrastructureIntent, cost: Optional[float], violations: List[str]) -> float:
+    """Risk from deployment scope (for config changes)."""
+    if not isinstance(intent, DeployConfigurationIntent):
+        return 0.0
+    mapping = {
+        "single_instance": 0.1,
+        "canary": 0.2,
+        "global": 0.6,
+    }
+    return mapping.get(intent.change_scope, 0.3)
+
+def environment_factor(intent: InfrastructureIntent, cost: Optional[float], violations: List[str]) -> float:
+    """Additional risk if environment is production."""
+    if hasattr(intent, "environment") and intent.environment == Environment.PROD:
+        return 0.1
+    return 0.0
+
+def policy_violation_factor(intent: InfrastructureIntent, cost: Optional[float], violations: List[str]) -> float:
+    """Risk from number of policy violations (capped)."""
+    # Each violation adds 0.2, max 0.8
+    return min(len(violations) * 0.2, 0.8)
+
+# -----------------------------------------------------------------------------
+# Risk Engine
+# -----------------------------------------------------------------------------
+class RiskEngine:
+    """
+    Computes a weighted risk score from multiple factors.
+
+    The engine is initialized with a list of factors and their weights.
+    The total score is the weighted sum of factor scores, clamped to [0,1].
+    """
+
+    DEFAULT_FACTORS = [
+        RiskFactor("intent_type", 1.0, intent_type_factor, "Base risk from intent type"),
+        RiskFactor("cost", 0.3, cost_factor, "Normalized cost estimate"),
+        RiskFactor("permission", 0.3, permission_factor, "Permission level being granted"),
+        RiskFactor("scope", 0.2, scope_factor, "Deployment scope"),
+        RiskFactor("environment", 0.1, environment_factor, "Production environment"),
+        RiskFactor("policy_violations", 0.2, policy_violation_factor, "Number of policy violations"),
+    ]
+
+    def __init__(self, factors: Optional[List[RiskFactor]] = None):
+        """
+        Initialize with custom factors. If none provided, uses DEFAULT_FACTORS.
+        """
+        self.factors = factors if factors is not None else self.DEFAULT_FACTORS
+
+    def calculate_risk(
+        self,
+        intent: InfrastructureIntent,
+        cost_estimate: Optional[float],
+        policy_violations: List[str],
+    ) -> Tuple[float, str, Dict[str, float]]:
+        """
+        Compute risk score and detailed breakdown.
+
+        Returns:
+            - total_score: float in [0,1]
+            - explanation: human-readable string
+            - contributions: dict mapping factor names to their weighted contribution
+        """
+        total = 0.0
+        contributions = {}
+
+        for factor in self.factors:
+            raw_score = factor(intent, cost_estimate, policy_violations)
+            weighted = raw_score * factor.weight
+            contributions[factor.name] = weighted
+            total += weighted
+
+        # Clamp to [0,1]
+        total = max(0.0, min(total, 1.0))
+
+        # Build explanation
+        lines = [f"Total risk score: {total:.2f}"]
+        for factor in self.factors:
+            contrib = contributions[factor.name]
+            if contrib > 0.0:
+                lines.append(f"  - {factor.name}: {contrib:.2f} ({factor.description})")
+        explanation = "\n".join(lines)
+
+        return total, explanation, contributions