Accountability & Governance

# ─── 1. AI Model Registry with Audit Trail ───────────────────
from dataclasses import dataclass, field
from datetime import datetime
from typing import List, Optional, Dict
from enum import Enum
import json

class ModelStatus(Enum):
    DEVELOPMENT = "development"
    REVIEW = "under_review"
    APPROVED = "approved"
    DEPLOYED = "deployed"
    DEPRECATED = "deprecated"
    RETIRED = "retired"

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

@dataclass
class AuditEvent:
    timestamp: str
    actor: str
    actor_role: str
    action: str
    details: str
    status_before: Optional[str] = None
    status_after: Optional[str] = None

@dataclass
class ModelRecord:
    model_id: str
    name: str
    version: str
    use_case: str
    risk_level: RiskLevel
    eu_ai_act_class: str
    owner: str          # Product Owner — Accountable for deployment
    developer: str      # ML Engineer — Responsible for technical quality
    data_steward: str   # Accountable for data quality
    status: ModelStatus = ModelStatus.DEVELOPMENT
    audit_trail: List[AuditEvent] = field(default_factory=list)
    impact_assessment_id: Optional[str] = None
    deployment_approver: Optional[str] = None
    deployment_date: Optional[str] = None
    monitoring_dashboard: Optional[str] = None

    def log_event(self, actor: str, role: str, action: str, details: str,
                   new_status: Optional[ModelStatus] = None):
        event = AuditEvent(
            timestamp=datetime.utcnow().isoformat() + "Z",
            actor=actor,
            actor_role=role,
            action=action,
            details=details,
            status_before=self.status.value,
            status_after=new_status.value if new_status else None,
        )
        if new_status:
            self.status = new_status
        self.audit_trail.append(event)
        return event

    def submit_for_review(self, submitter: str, impact_assessment_id: str):
        self.impact_assessment_id = impact_assessment_id
        return self.log_event(
            submitter, "ML Engineer",
            "SUBMIT_FOR_REVIEW",
            f"Submitted with impact assessment {impact_assessment_id}",
            ModelStatus.REVIEW,
        )

    def approve_deployment(self, approver: str, notes: str = ""):
        self.deployment_approver = approver
        self.deployment_date = datetime.utcnow().isoformat() + "Z"
        return self.log_event(
            approver, "CTO/CPO",
            "APPROVE_DEPLOYMENT",
            f"Approved for production. {notes}",
            ModelStatus.APPROVED,
        )

    def reject(self, reviewer: str, reason: str):
        return self.log_event(
            reviewer, "AI Ethics Board",
            "REJECT",
            f"Rejected: {reason}",
            ModelStatus.DEVELOPMENT,
        )

    def deploy(self, deployer: str, dashboard_url: str):
        self.monitoring_dashboard = dashboard_url
        return self.log_event(
            deployer, "DevOps",
            "DEPLOY",
            f"Deployed to production. Dashboard: {dashboard_url}",
            ModelStatus.DEPLOYED,
        )

    def print_audit_trail(self):
        print(f"\nAudit Trail — {self.name} v{self.version}")
        print(f"{'─'*70}")
        for event in self.audit_trail:
            print(f"  [{event.timestamp}] {event.actor} ({event.actor_role})")
            print(f"    Action: {event.action}")
            print(f"    Details: {event.details}")
            if event.status_before:
                print(f"    Status: {event.status_before} → {event.status_after}")

# Simulate a model governance lifecycle
loan_model = ModelRecord(
    model_id="LM-2025-001",
    name="LoanApprovalAI",
    version="3.2.0",
    use_case="Consumer loan application scoring",
    risk_level=RiskLevel.HIGH,
    eu_ai_act_class="HIGH RISK — financial services",
    owner="Sarah Chen (Product)",
    developer="Alex Kumar (ML)",
    data_steward="Maria Santos (Data)",
)

# Log the development lifecycle
loan_model.log_event("alex.kumar", "ML Engineer", "MODEL_TRAINING",
    "Trained XGBoost model on 250k samples, accuracy=87.3%, AUC=0.92")
loan_model.log_event("alex.kumar", "ML Engineer", "FAIRNESS_TEST",
    "Disparate Impact Ratio: 0.91 (PASS). DPD: 0.03 (ACCEPTABLE). Certified by AI Ethics Board member.")
loan_model.log_event("maria.santos", "Data Steward", "DATA_QUALITY_SIGN_OFF",
    "Data lineage verified. No PII leakage. GDPR consent documented for all training samples.")
loan_model.submit_for_review("alex.kumar", "AIIA-2025-007")
loan_model.log_event("ethics.board", "AI Ethics Board", "ETHICS_REVIEW",
    "Reviewed AIIA-2025-007. Approved with condition: SHAP explanations must be provided for all rejections.")
loan_model.approve_deployment("cto@company.com", "Conditional: adverse action notices required.")
loan_model.deploy("devops.team", "https://grafana.company.com/d/loan-model-v3")

loan_model.print_audit_trail()

# ─── 2. AI Incident Response System ──────────────────────────
class IncidentSeverity(Enum):
    P1_CRITICAL = "P1"   # Halt deployment; immediate escalation to C-suite
    P2_HIGH = "P2"       # 24h response; AI Ethics Board notification
    P3_MEDIUM = "P3"     # 72h response; team lead notification
    P4_LOW = "P4"        # 1-week response; logged for audit

@dataclass
class AIIncident:
    incident_id: str
    model_id: str
    detected_at: str
    reported_by: str
    description: str
    severity: IncidentSeverity
    affected_users: int
    resolution_status: str = "OPEN"
    remediation_steps: List[str] = field(default_factory=list)
    root_cause: Optional[str] = None
    post_mortem_date: Optional[str] = None

    def classify_sla(self) -> str:
        sla_map = {
            IncidentSeverity.P1_CRITICAL: "Acknowledge within 15min; resolve within 4h; halt model",
            IncidentSeverity.P2_HIGH: "Acknowledge within 1h; resolve within 24h",
            IncidentSeverity.P3_MEDIUM: "Acknowledge within 4h; resolve within 72h",
            IncidentSeverity.P4_LOW: "Acknowledge within 24h; resolve within 7 days",
        }
        return sla_map[self.severity]

    def add_remediation(self, step: str):
        self.remediation_steps.append(step)

    def close(self, root_cause: str, post_mortem_date: str):
        self.root_cause = root_cause
        self.post_mortem_date = post_mortem_date
        self.resolution_status = "CLOSED"

    def to_report(self) -> dict:
        return {
            "incident_id": self.incident_id,
            "model": self.model_id,
            "severity": self.severity.value,
            "sla": self.classify_sla(),
            "affected_users": self.affected_users,
            "status": self.resolution_status,
            "remediation": self.remediation_steps,
            "root_cause": self.root_cause,
        }

# Example incident
incident = AIIncident(
    incident_id="INC-2025-0142",
    model_id="LM-2025-001",
    detected_at="2025-06-15T14:32:00Z",
    reported_by="monitoring.alert@company.com",
    description="Fairness monitoring alert: Disparate Impact Ratio dropped to 0.71 for Hispanic applicants over past 7 days — below legal threshold of 0.8",
    severity=IncidentSeverity.P1_CRITICAL,
    affected_users=2847,
)

incident.add_remediation("IMMEDIATE: Model rolled back to LM-2025-001 v3.1.0 (14:45 UTC)")
incident.add_remediation("INVESTIGATION: Data team analyzing input drift in zip code distribution")
incident.add_remediation("LEGAL: Compliance team notified; affected applicants flagged for manual review")
incident.add_remediation("ROOT CAUSE ANALYSIS: New data pipeline introduced geographic recoding error")
incident.close(
    root_cause="ETL pipeline bug recoded zip codes incorrectly after postcode database update, introducing measurement bias in the geographic features correlated with ethnicity.",
    post_mortem_date="2025-06-22",
)

print("\nAI Incident Report:")
print(json.dumps(incident.to_report(), indent=2))

# ─── 3. Governance Scorecard ──────────────────────────────────
governance_domains = {
    "Policy & Strategy": {
        "AI policy documented": True,
        "AI policy reviewed in past 12 months": True,
        "Board-level AI risk oversight": False,
        "AI Ethics Board established": True,
    },
    "Risk Management": {
        "AI Impact Assessments for all high-risk systems": True,
        "Third-party AI vendor assessments": False,
        "AI incident tracking system": True,
        "Post-incident root cause analysis process": True,
    },
    "Transparency": {
        "Model cards for all production models": False,
        "Datasheets for training datasets": False,
        "User-facing AI disclosure notices": True,
        "Appeals/redress process for AI decisions": False,
    },
    "Monitoring": {
        "Automated drift detection in production": True,
        "Fairness monitoring dashboards": True,
        "Quarterly model performance reviews": True,
        "Annual third-party AI audit": False,
    },
}

print("\nAI Governance Scorecard:")
total, implemented = 0, 0
for domain, controls in governance_domains.items():
    domain_yes = sum(v for v in controls.values())
    domain_total = len(controls)
    pct = 100 * domain_yes // domain_total
    print(f"\n  {domain} ({domain_yes}/{domain_total} = {pct}%)")
    for ctrl, status in controls.items():
        print(f"    {'✓' if status else '✗'} {ctrl}")
    total += domain_total
    implemented += domain_yes

print(f"\nOverall Governance Score: {implemented}/{total} = {100*implemented//total}%")