Implementing Enterprise AI Governance: Code Examples and Deployment Patterns

from typing import Dict, List, Any, Optional
from dataclasses import dataclass, field
from datetime import datetime, timedelta
import asyncio
import json
from enum import Enum

class RiskLevel(Enum):
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"
    CRITICAL = "critical"

@dataclass
class RiskDimension:
    name: str
    weight: float
    score: float
    factors: Dict[str, Any] = field(default_factory=dict)
    justification: str = ""

@dataclass
class RiskAssessment:
    initiative_id: str
    overall_risk_level: RiskLevel
    overall_risk_score: float
    risk_dimensions: List[RiskDimension]
    recommended_pathway: str
    assessment_timestamp: datetime
    assessor_metadata: Dict[str, Any] = field(default_factory=dict)

class EnterpriseRiskEngine:
    """Production-ready risk assessment engine for AI governance."""

    def __init__(self, config_manager, policy_engine, data_classifier):
        self.config = config_manager
        self.policy_engine = policy_engine
        self.data_classifier = data_classifier
        self.risk_models = self._load_risk_models()

    async def assess_initiative(self, initiative_metadata: Dict[str, Any]) -> RiskAssessment:
        """Perform comprehensive risk assessment on AI initiative."""

        try:
            # Parallel risk dimension assessment
            risk_assessment_tasks = [
                self._assess_technical_risk(initiative_metadata),
                self._assess_business_risk(initiative_metadata),
                self._assess_regulatory_risk(initiative_metadata),
                self._assess_ethical_risk(initiative_metadata),
                self._assess_operational_risk(initiative_metadata)
            ]

            risk_dimensions = await asyncio.gather(*risk_assessment_tasks)

            # Calculate overall risk score
            overall_score = self._calculate_weighted_risk_score(risk_dimensions)
            risk_level = self._determine_risk_level(overall_score)

            # Determine approval pathway
            approval_pathway = await self._determine_approval_pathway(
                risk_level, risk_dimensions, initiative_metadata
            )

            assessment = RiskAssessment(
                initiative_id=initiative_metadata["id"],
                overall_risk_level=risk_level,
                overall_risk_score=overall_score,
                risk_dimensions=risk_dimensions,
                recommended_pathway=approval_pathway,
                assessment_timestamp=datetime.utcnow(),
                assessor_metadata={
                    "assessment_version": "1.2.0",
                    "model_versions": self._get_model_versions(),
                    "policy_version": await self.policy_engine.get_current_version()
                }
            )

            # Store assessment for audit trail
            await self._store_assessment(assessment)

            return assessment

        except Exception as e:
            # Log error and return safe default
            await self._log_assessment_error(initiative_metadata["id"], str(e))
            return self._create_default_high_risk_assessment(initiative_metadata["id"])

    async def _assess_technical_risk(self, metadata: Dict[str, Any]) -> RiskDimension:
        """Assess technical risk factors."""

        technical_factors = {}
        technical_score = 0.0

        # Data sensitivity assessment
        data_sources = metadata.get("data_sources", [])
        data_sensitivity = await self._assess_data_sensitivity(data_sources)
        technical_factors["data_sensitivity"] = data_sensitivity
        technical_score += data_sensitivity["score"] * 0.3

        # Model complexity assessment
        model_type = metadata.get("model_type", "unknown")
        model_complexity = await self._assess_model_complexity(model_type, metadata)
        technical_factors["model_complexity"] = model_complexity
        technical_score += model_complexity["score"] * 0.25

        # Integration complexity
        integration_points = metadata.get("integration_points", [])
        integration_risk = await self._assess_integration_complexity(integration_points)
        technical_factors["integration_complexity"] = integration_risk
        technical_score += integration_risk["score"] * 0.2

        # Security considerations
        security_requirements = metadata.get("security_requirements", {})
        security_risk = await self._assess_security_risk(security_requirements)
        technical_factors["security_risk"] = security_risk
        technical_score += security_risk["score"] * 0.25

        return RiskDimension(
            name="technical_risk",
            weight=0.25,
            score=min(technical_score, 1.0),
            factors=technical_factors,
            justification=self._generate_technical_risk_justification(technical_factors)
        )

    async def _assess_data_sensitivity(self, data_sources: List[Dict]) -> Dict[str, Any]:
        """Assess sensitivity of data sources using enterprise data classifier."""

        max_sensitivity = 0.0
        sensitivity_details = {}

        for data_source in data_sources:
            classification = await self.data_classifier.classify_data_source(
                data_source["location"],
                data_source.get("schema", {}),
                data_source.get("sample_data", {})
            )

            sensitivity_score = self._map_classification_to_risk(classification)
            max_sensitivity = max(max_sensitivity, sensitivity_score)

            sensitivity_details[data_source["name"]] = {
                "classification": classification,
                "sensitivity_score": sensitivity_score,
                "contains_pii": classification.get("contains_pii", False),
                "regulatory_scope": classification.get("regulatory_scope", [])
            }

        return {
            "score": max_sensitivity,
            "details": sensitivity_details,
            "highest_classification": max(
                [d["classification"]["level"] for d in sensitivity_details.values()]
            )
        }

    async def _assess_business_risk(self, metadata: Dict[str, Any]) -> RiskDimension:
        """Assess business impact and risk factors."""

        business_factors = {}
        business_score = 0.0

        # Customer impact assessment
        customer_impact = metadata.get("customer_impact", {})
        customer_risk = self._calculate_customer_impact_risk(customer_impact)
        business_factors["customer_impact"] = customer_risk
        business_score += customer_risk["score"] * 0.4

        # Financial impact assessment
        financial_impact = metadata.get("financial_impact", {})
        financial_risk = self._calculate_financial_risk(financial_impact)
        business_factors["financial_impact"] = financial_risk
        business_score += financial_risk["score"] * 0.3

        # Competitive sensitivity
        competitive_sensitivity = metadata.get("competitive_sensitivity", "low")
        competitive_risk = self._assess_competitive_sensitivity(competitive_sensitivity)
        business_factors["competitive_sensitivity"] = competitive_risk
        business_score += competitive_risk["score"] * 0.3

        return RiskDimension(
            name="business_risk",
            weight=0.3,
            score=min(business_score, 1.0),
            factors=business_factors,
            justification=self._generate_business_risk_justification(business_factors)
        )

    def _calculate_weighted_risk_score(self, risk_dimensions: List[RiskDimension]) -> float:
        """Calculate overall weighted risk score."""

        total_weighted_score = 0.0
        total_weight = 0.0

        for dimension in risk_dimensions:
            total_weighted_score += dimension.score * dimension.weight
            total_weight += dimension.weight

        return total_weighted_score / total_weight if total_weight > 0 else 0.0

    def _determine_risk_level(self, risk_score: float) -> RiskLevel:
        """Map risk score to risk level based on enterprise thresholds."""

        thresholds = self.config.get_risk_thresholds()

        if risk_score >= thresholds["critical"]:
            return RiskLevel.CRITICAL
        elif risk_score >= thresholds["high"]:
            return RiskLevel.HIGH
        elif risk_score >= thresholds["medium"]:
            return RiskLevel.MEDIUM
        else:
            return RiskLevel.LOW

from typing import Dict, List, Set, Optional
from enum import Enum
from dataclasses import dataclass, field
import asyncio
from datetime import datetime, timedelta

class WorkflowState(Enum):
    INITIATED = "initiated"
    RISK_ASSESSED = "risk_assessed"
    TECHNICAL_REVIEW = "technical_review"
    BUSINESS_REVIEW = "business_review"
    COMPLIANCE_REVIEW = "compliance_review"
    EXECUTIVE_REVIEW = "executive_review"
    APPROVED = "approved"
    CONDITIONAL_APPROVAL = "conditional_approval"
    REJECTED = "rejected"
    ESCALATED = "escalated"

@dataclass
class ApprovalStep:
    step_id: str
    approver_role: str
    required_approvers: Set[str]
    optional_approvers: Set[str] = field(default_factory=set)
    sla_hours: int = 24
    auto_approve_conditions: List[str] = field(default_factory=list)
    escalation_rules: Dict[str, Any] = field(default_factory=dict)

@dataclass
class WorkflowInstance:
    workflow_id: str
    initiative_id: str
    current_state: WorkflowState
    approval_pathway: str
    steps: List[ApprovalStep]
    completed_steps: List[Dict[str, Any]] = field(default_factory=list)
    pending_approvals: Set[str] = field(default_factory=set)
    conditions: List[str] = field(default_factory=list)
    created_at: datetime = field(default_factory=datetime.utcnow)
    sla_deadline: Optional[datetime] = None

class WorkflowEngine:
    """Enterprise workflow engine for AI governance approval processes."""

    def __init__(self, notification_service, user_service, audit_service):
        self.notification_service = notification_service
        self.user_service = user_service
        self.audit_service = audit_service
        self.workflow_definitions = self._load_workflow_definitions()
        self.active_workflows: Dict[str, WorkflowInstance] = {}

    async def initiate_workflow(
        self,
        initiative_id: str,
        risk_assessment: RiskAssessment,
        requester_id: str
    ) -> WorkflowInstance:
        """Initiate approval workflow based on risk assessment."""

        workflow_id = self._generate_workflow_id()
        pathway = risk_assessment.recommended_pathway

        # Get workflow definition
        workflow_definition = self.workflow_definitions[pathway]

        # Create workflow instance
        workflow = WorkflowInstance(
            workflow_id=workflow_id,
            initiative_id=initiative_id,
            current_state=WorkflowState.INITIATED,
            approval_pathway=pathway,
            steps=workflow_definition["steps"],
            sla_deadline=datetime.utcnow() + timedelta(
                hours=workflow_definition["total_sla_hours"]
            )
        )

        # Store workflow
        self.active_workflows[workflow_id] = workflow

        # Transition to first approval step
        await self._transition_to_next_step(workflow)

        # Send notifications
        await self._send_workflow_initiation_notifications(workflow, requester_id)

        return workflow

    async def process_approval_decision(
        self,
        workflow_id: str,
        approver_id: str,
        decision: str,
        comments: str = "",
        conditions: List[str] = None
    ) -> WorkflowInstance:
        """Process an approval decision and advance workflow."""

        workflow = self.active_workflows.get(workflow_id)
        if not workflow:
            raise ValueError(f"Workflow {workflow_id} not found")

        # Validate approver authority
        await self._validate_approver_authority(workflow, approver_id)

        # Record approval decision
        approval_record = {
            "approver_id": approver_id,
            "decision": decision,
            "comments": comments,
            "conditions": conditions or [],
            "timestamp": datetime.utcnow(),
            "step_id": self._get_current_step_id(workflow)
        }

        workflow.completed_steps.append(approval_record)

        # Update workflow conditions
        if conditions:
            workflow.conditions.extend(conditions)

        # Process decision
        if decision == "approved":
            await self._handle_approval(workflow, approval_record)
        elif decision == "rejected":
            await self._handle_rejection(workflow, approval_record)
        elif decision == "conditional_approval":
            await self._handle_conditional_approval(workflow, approval_record)
        elif decision == "escalate":
            await self._handle_escalation(workflow, approval_record)

        # Audit decision
        await self.audit_service.log_approval_decision(workflow_id, approval_record)

        return workflow

    async def _handle_approval(self, workflow: WorkflowInstance, approval_record: Dict):
        """Handle approval decision and potentially advance workflow."""

        current_step = self._get_current_step(workflow)

        # Remove approver from pending list
        workflow.pending_approvals.discard(approval_record["approver_id"])

        # Check if all required approvers have approved
        required_approvals = current_step.required_approvers
        completed_approvers = {
            step["approver_id"] for step in workflow.completed_steps
            if step["decision"] == "approved"
        }

        if required_approvals.issubset(completed_approvers):
            # Step complete, advance workflow
            await self._transition_to_next_step(workflow)
        else:
            # Still waiting for more approvals
            await self._send_pending_approval_notifications(workflow)

    async def _transition_to_next_step(self, workflow: WorkflowInstance):
        """Transition workflow to next approval step."""

        current_step_index = self._get_current_step_index(workflow)

        if current_step_index < len(workflow.steps) - 1:
            # Move to next step
            next_step = workflow.steps[current_step_index + 1]
            workflow.current_state = self._map_step_to_state(next_step)

            # Set up pending approvals
            workflow.pending_approvals = next_step.required_approvers.copy()

            # Send notifications to new approvers
            await self._send_step_notifications(workflow, next_step)

            # Set up auto-approval check if configured
            if next_step.auto_approve_conditions:
                await self._check_auto_approval_conditions(workflow, next_step)

        else:
            # Workflow complete
            workflow.current_state = WorkflowState.APPROVED
            await self._handle_workflow_completion(workflow)

    async def _check_auto_approval_conditions(
        self, workflow: WorkflowInstance, step: ApprovalStep
    ):
        """Check if auto-approval conditions are met."""

        for condition in step.auto_approve_conditions:
            if await self._evaluate_auto_approval_condition(workflow, condition):
                # Auto-approve this step
                auto_approval_record = {
                    "approver_id": "system",
                    "decision": "approved",
                    "comments": f"Auto-approved based on condition: {condition}",
                    "conditions": [],
                    "timestamp": datetime.utcnow(),
                    "step_id": step.step_id
                }

                workflow.completed_steps.append(auto_approval_record)
                await self._transition_to_next_step(workflow)
                break

    async def check_sla_violations(self):
        """Check for SLA violations and trigger escalations."""

        current_time = datetime.utcnow()

        for workflow in self.active_workflows.values():
            if workflow.sla_deadline and current_time > workflow.sla_deadline:
                await self._handle_sla_violation(workflow)

    async def _handle_sla_violation(self, workflow: WorkflowInstance):
        """Handle SLA violation with appropriate escalation."""

        current_step = self._get_current_step(workflow)
        escalation_rules = current_step.escalation_rules

        violation_hours = (datetime.utcnow() - workflow.sla_deadline).total_seconds() / 3600

        if violation_hours <= escalation_rules.get("auto_approve_threshold_hours", 24):
            # Auto-approve with enhanced monitoring
            await self._auto_approve_with_monitoring(workflow)
        else:
            # Escalate to executive review
            await self._escalate_to_executive(workflow)

import json
from typing import Dict, Any, List, Optional
from dataclasses import dataclass
from datetime import datetime

@dataclass
class PolicyRule:
    rule_id: str
    name: str
    condition: str
    action: str
    priority: int
    effective_date: datetime
    expiry_date: Optional[datetime] = None
    exceptions: List[str] = None

class PolicyEngine:
    """Enterprise policy engine for AI governance decisions."""

    def __init__(self, policy_repository, rule_evaluator):
        self.policy_repository = policy_repository
        self.rule_evaluator = rule_evaluator
        self.active_policies = {}
        self._load_active_policies()

    async def evaluate_policies(
        self,
        initiative_metadata: Dict[str, Any],
        risk_scores: List[float]
    ) -> Dict[str, Any]:
        """Evaluate all applicable policies against initiative."""

        applicable_policies = await self._get_applicable_policies(initiative_metadata)
        policy_results = []

        for policy in applicable_policies:
            try:
                result = await self._evaluate_single_policy(
                    policy, initiative_metadata, risk_scores
                )
                policy_results.append(result)
            except Exception as e:
                # Log policy evaluation error
                await self._log_policy_error(policy.rule_id, str(e))
                continue

        # Resolve policy conflicts
        final_decision = await self._resolve_policy_conflicts(policy_results)

        return {
            "decision": final_decision["action"],
            "applicable_policies": [p.rule_id for p in applicable_policies],
            "policy_results": policy_results,
            "conflict_resolution": final_decision.get("conflict_resolution"),
            "evaluation_timestamp": datetime.utcnow().isoformat()
        }

    async def _evaluate_single_policy(
        self,
        policy: PolicyRule,
        metadata: Dict[str, Any],
        risk_scores: List[float]
    ) -> Dict[str, Any]:
        """Evaluate a single policy rule."""

        # Prepare evaluation context
        context = {
            "metadata": metadata,
            "risk_scores": risk_scores,
            "overall_risk": max(risk_scores) if risk_scores else 0.0,
            "current_time": datetime.utcnow(),
            "business_hours": self._is_business_hours(),
            "emergency_mode": metadata.get("emergency_mode", False)
        }

        # Evaluate condition
        condition_result = await self.rule_evaluator.evaluate_condition(
            policy.condition, context
        )

        # Check for exceptions
        exception_applied = await self._check_exceptions(policy, context)

        # Determine action
        if condition_result and not exception_applied:
            action = policy.action
            applies = True
        else:
            action = "continue"
            applies = False

        return {
            "policy_id": policy.rule_id,
            "policy_name": policy.name,
            "applies": applies,
            "action": action,
            "condition_result": condition_result,
            "exception_applied": exception_applied,
            "priority": policy.priority
        }

    async def _resolve_policy_conflicts(
        self, policy_results: List[Dict[str, Any]]
    ) -> Dict[str, Any]:
        """Resolve conflicts between multiple applicable policies."""

        applicable_results = [r for r in policy_results if r["applies"]]

        if not applicable_results:
            return {"action": "default_approval_pathway"}

        # Sort by priority (higher number = higher priority)
        applicable_results.sort(key=lambda x: x["priority"], reverse=True)

        # Check for conflicts
        actions = [r["action"] for r in applicable_results]
        unique_actions = set(actions)

        if len(unique_actions) == 1:
            # No conflict
            return {
                "action": actions[0],
                "conflict_resolution": "no_conflict"
            }
        else:
            # Resolve conflict using priority
            highest_priority_result = applicable_results[0]
            return {
                "action": highest_priority_result["action"],
                "conflict_resolution": "priority_based",
                "conflicting_policies": [r["policy_id"] for r in applicable_results],
                "winning_policy": highest_priority_result["policy_id"]
            }

import logging
from prometheus_client import Counter, Histogram, Gauge
from datadog import initialize, statsd
import structlog

# Metrics collection
RISK_ASSESSMENTS_TOTAL = Counter(
    'governance_risk_assessments_total',
    'Total number of risk assessments performed',
    ['risk_level', 'pathway']
)

APPROVAL_PROCESSING_TIME = Histogram(
    'governance_approval_processing_seconds',
    'Time spent processing approvals',
    ['pathway', 'outcome']
)

ACTIVE_WORKFLOWS = Gauge(
    'governance_active_workflows',
    'Number of active approval workflows',
    ['state']
)

SLA_VIOLATIONS = Counter(
    'governance_sla_violations_total',
    'Total number of SLA violations',
    ['violation_type']
)

class GovernanceMonitoring:
    """Comprehensive monitoring for AI governance systems."""

    def __init__(self, config):
        self.config = config
        self.logger = structlog.get_logger()

        # Initialize DataDog if configured
        if config.get("datadog_api_key"):
            initialize(
                api_key=config["datadog_api_key"],
                app_key=config["datadog_app_key"]
            )

    def record_risk_assessment(self, assessment: RiskAssessment):
        """Record metrics for risk assessment."""

        RISK_ASSESSMENTS_TOTAL.labels(
            risk_level=assessment.overall_risk_level.value,
            pathway=assessment.recommended_pathway
        ).inc()

        # Send to DataDog
        statsd.increment(
            'governance.risk_assessment.completed',
            tags=[
                f'risk_level:{assessment.overall_risk_level.value}',
                f'pathway:{assessment.recommended_pathway}'
            ]
        )

        # Structured logging
        self.logger.info(
            "risk_assessment_completed",
            initiative_id=assessment.initiative_id,
            risk_level=assessment.overall_risk_level.value,
            risk_score=assessment.overall_risk_score,
            pathway=assessment.recommended_pathway
        )

    def record_approval_decision(self, workflow: WorkflowInstance, decision_time: float):
        """Record metrics for approval decisions."""

        APPROVAL_PROCESSING_TIME.labels(
            pathway=workflow.approval_pathway,
            outcome=workflow.current_state.value
        ).observe(decision_time)

        statsd.histogram(
            'governance.approval.processing_time',
            decision_time,
            tags=[
                f'pathway:{workflow.approval_pathway}',
                f'outcome:{workflow.current_state.value}'
            ]
        )

    def record_sla_violation(self, workflow: WorkflowInstance, violation_type: str):
        """Record SLA violation metrics."""

        SLA_VIOLATIONS.labels(violation_type=violation_type).inc()

        statsd.increment(
            'governance.sla.violation',
            tags=[
                f'violation_type:{violation_type}',
                f'pathway:{workflow.approval_pathway}'
            ]
        )

        self.logger.warning(
            "sla_violation_detected",
            workflow_id=workflow.workflow_id,
            violation_type=violation_type,
            pathway=workflow.approval_pathway,
            elapsed_hours=(datetime.utcnow() - workflow.created_at).total_seconds() / 3600
        )