# Compliance

Regulatory compliance for agent systems: GDPR, SOC2, HIPAA, PCI-DSS, AI regulations, and audit readiness.

---

## Compliance Framework

```
Compliance Program
├── Governance
│   ├── Policy Management
│   ├── Risk Assessment
│   ├── Control Framework
│   └── Continuous Monitoring
├── Regulatory Requirements
│   ├── Data Protection (GDPR, CCPA, HIPAA)
│   ├── Financial (PCI-DSS, SOX)
│   ├── AI-Specific (EU AI Act, NIST AI RMF)
│   └── Industry-Specific (FedRAMP, StateRAMP)
├── Technical Controls
│   ├── Data Protection
│   ├── Access Control
│   ├── Audit Logging
│   ├── Encryption
│   └── Vulnerability Management
├── Operational
│   ├── Incident Response
│   ├── Vendor Management
│   ├── Training & Awareness
│   └── Audit & Assessment
└── Documentation
    ├── Policies & Procedures
    ├── Evidence Collection
    ├── Audit Trails
    └── Compliance Reporting
```

---

## Data Protection Regulations

### GDPR Compliance
```python
class GDPRCompliance:
    def __init__(self, encryption: EncryptionService, audit_logger: AuditLogger):
        self.encryption = encryption
        self.audit = audit_logger
        self.retention_policies = RetentionPolicyManager()
    
    # Article 5: Lawfulness, fairness, transparency
    async def process_lawfully(self, processing: DataProcessing) -> LawfulBasis:
        """Determine lawful basis for processing"""
        bases = {
            "consent": await self.check_consent(processing),
            "contract": await self.check_contractual_necessity(processing),
            "legal_obligation": await self.check_legal_obligation(processing),
            "vital_interests": await self.check_vital_interests(processing),
            "public_task": await self.check_public_task(processing),
            "legitimate_interests": await self.assess_legitimate_interests(processing)
        )
        
        valid = [b for b, valid in bases.items() if valid]
        if not valid:
            raise GDPRViolation("No lawful basis for processing")
        
        return LawfulBasis(primary=valid[0], all_bases=valid)
    
    # Article 12-14: Transparency & Information
    async def provide_privacy_notice(self, data_subject: DataSubject) -> PrivacyNotice:
        return PrivacyNotice(
            controller=CONTROLLER_INFO,
            purposes=self.get_processing_purposes(),
            lawful_basis=await self.get_lawful_basis(),
            recipients=self.get_recipients(),
            transfers=self.get_international_transfers(),
            retention=self.get_retention_periods(),
            rights=self.get_data_subject_rights(),
            contact=DPO_CONTACT
        )
    
    # Article 15: Right of Access
    async def handle_access_request(self, request: AccessRequest) -> AccessResponse:
        # Verify identity
        await self.verify_identity(request.data_subject)
        
        # Collect data
        data = await self.collect_personal_data(request.data_subject)
        
        # Log request
        await self.audit.log(AccessLog(
            data_subject=request.data_subject,
            request_type="access",
            data_categories=self.categorize_data(data),
            provided=True
        ))
        
        return AccessResponse(data=data)
    
    # Article 17: Right to Erasure
    async def handle_erasure_request(self, request: ErasureRequest) -> ErasureResponse:
        # Check if erasure applies
        if not await self.erasure_applies(request):
            return ErasureResponse(granted=False, reason="Erasure not applicable")
        
        # Execute erasure
        result = await self.execute_erasure(request.data_subject)
        
        # Log
        await self.audit.log(AccessLog(
            data_subject=request.data_subject,
            request_type="erasure",
            result="completed",
            records_affected=result.records_deleted
        ))
        
        return ErasureResponse(granted=True, records_deleted=result.records_deleted)
    
    # Article 20: Data Portability
    async def handle_portability_request(self, request: PortabilityRequest) -> PortabilityResponse:
        data = await self.export_portable_data(request.data_subject)
        return PortabilityResponse(
            data=data,
            format="json",
            includes=self.get_portable_categories()
        )
    
    # Article 25: Data Protection by Design & Default
    async def privacy_impact_assessment(self, processing: DataProcessing) -> DPIA:
        """Data Protection Impact Assessment"""
        risks = await self.identify_risks(processing)
        mitigations = await self.design_mitigations(risks)
        
        return DPIA(
            processing=processing,
            risks=risks,
            mitigations=mitigations,
            residual_risk=self.assess_residual_risk(risks, mitigations),
            dpo_consulted=await self.consult_dpo(processing)
        )
    
    # Article 32: Security of Processing
    async def implement_security_measures(self) -> SecurityMeasures:
        return SecurityMeasures(
            encryption=self.encryption.get_status(),
            access_control=await self.audit_access_controls(),
            pseudonymization=self.pseudonymization_status(),
            resilience=self.test_resilience(),
            restore_time=self.measure_restore_time(),
            testing=self.schedule_security_testing()
        )
    
    # Article 33-34: Breach Notification
    async def handle_breach(self, breach: DataBreach) -> BreachResponse:
        # Assess risk
        risk = await self.assess_breach_risk(breach)
        
        # Notify supervisory authority (72 hours)
        if risk.level in [RiskLevel.HIGH, RiskLevel.CRITICAL]:
            await self.notify_authority(breach, within_hours=72)
        
        # Notify data subjects (without undue delay)
        if risk.level == RiskLevel.CRITICAL:
            await self.notify_data_subjects(breach)
        
        # Document
        await self.document_breach(breach)
        
        return BreachResponse(notified_authority=risk.level in [RiskLevel.HIGH, RiskLevel.CRITICAL],
                              notified_subjects=risk.level == RiskLevel.CRITICAL)
```

### CCPA/CPRA Compliance
```python
class CCPACompliance:
    def __init__(self):
        self.consumer_requests = ConsumerRequestManager()
        self.sale_tracker = SaleTracker()
    
    async def handle_consumer_request(self, request: ConsumerRequest) -> ConsumerResponse:
        if request.type == RequestType.KNOW:
            return await self.handle_know_request(request)
        elif request.type == RequestType.DELETE:
            return await self.handle_delete_request(request)
        elif request.type == RequestType.OPT_OUT:
            return await self.handle_opt_out_request(request)
        elif request.type == RequestType.LIMIT_USE:
            return await self.handle_limit_use_request(request)
    
    async def handle_know_request(self, request: KnowRequest) -> KnowResponse:
        # Categories of personal information collected
        categories = await self.get_collection_categories(request.consumer)
        
        # Specific pieces of personal information
        pieces = await self.get_specific_pieces(request.consumer)
        
        # Sources
        sources = await self.get_sources(request.consumer)
        
        # Business purpose
        purposes = await self.get_business_purposes(request.consumer)
        
        # Third parties
        third_parties = await self.get_third_parties(request.consumer)
        
        return KnowResponse(
            categories=categories,
            pieces=pieces,
            sources=sources,
            purposes=purposes,
            third_parties=third_parties
        )
    
    async def handle_opt_out_request(self, request: OptOutRequest) -> OptOutResponse:
        # Stop sale/sharing
        await self.sale_tracker.stop_sale(request.consumer)
        
        # Notify downstream
        await self.notify_downstream(request.consumer)
        
        return OptOutResponse(success=True, effective_date=datetime.utcnow())
```

---

## AI-Specific Regulations

### EU AI Act Compliance
```python
class EUAIActCompliance:
    def __init__(self):
        self.risk_classifier = AIRiskClassifier()
        self.conformity = ConformityAssessment()
    
    async def classify_system(self, system: AISystem) -> RiskClassification:
        """Classify AI system per EU AI Act"""
        # Unacceptable risk
        if self.is_unacceptable_risk(system):
            return RiskClassification.UNACCEPTABLE
        
        # High risk
        if self.is_high_risk(system):
            return RiskClassification.HIGH_RISK
        
        # Limited risk
        if self.is_limited_risk(system):
            return RiskClassification.LIMITED
        
        # Minimal risk
        return RiskClassification.MINIMAL
    
    def is_high_risk(self, system: AISystem) -> bool:
        # Annex III categories
        high_risk_categories = [
            "biometric_identification",
            "critical_infrastructure",
            "education_vocational_training",
            "employment_worker_management",
            "essential_services",
            "law_enforcement",
            "migration_asylum_border",
            "administration_of_justice"
        ]
        
        return system.domain in high_risk_categories
    
    async def compliance_requirements(self, system: AISystem) -> ComplianceRequirements:
        classification = await self.classify_system(system)
        
        if classification == RiskClassification.UNACCEPTABLE:
            return ComplianceRequirements(
                allowed=False,
                reason="Unacceptable risk under EU AI Act"
            )
        
        if classification == RiskClassification.HIGH_RISK:
            return ComplianceRequirements(
                requirements=[
                    "risk_management_system",
                    "data_governance",
                    "technical_documentation",
                    "record_keeping",
                    "transparency_user_info",
                    "human_oversight",
                    "accuracy_robustness_cybersecurity",
                    "conformity_assessment",
                    "ce_marking",
                    "registration"
                ],
                conformity_assessment=await self.conformity.assess(system)
            )
        
        return ComplianceRequirements(
            requirements=["transparency_obligations"] if classification == RiskClassification.LIMITED else []
        )
```

### NIST AI Risk Management Framework
```python
class NISTAIRMFCompliance:
    def __init__(self):
        self.functions = {
            "govern": GovernFunction(),
            "map": MapFunction(),
            "measure": MeasureFunction(),
            "manage": ManageFunction()
        }
    
    async def assess(self, system: AISystem) -> NISTAssessment:
        results = {}
        for name, func in self.functions.items():
            results[name] = await func.assess(system)
        
        return NISTAssessment(
            govern=results["govern"],
            map=results["map"],
            measure=results["measure"],
            manage=results["manage"],
            overall_score=self.calculate_score(results)
        )

class GovernFunction:
    async def assess(self, system: AISystem) -> GovernResult:
        return GovernResult(
            policies_exist=self.check_policies(system),
            roles_defined=self.check_roles(system),
            accountability=self.check_accountability(system),
            culture=self.assess_culture(system)
        )

class MapFunction:
    async def assess(self, system: AISystem) -> MapResult:
        return MapResult(
            context_documented=self.check_context(system),
            stakeholders_identified=self.check_stakeholders(system),
            risks_identified=self.identify_risks(system),
            benefits_documented=self.check_benefits(system)
        )

class MeasureFunction:
    async def assess(self, system: AISystem) -> MeasureResult:
        return MeasureResult(
            metrics_defined=self.check_metrics(system),
            testing_performed=self.check_testing(system),
            benchmarks_established=self.check_benchmarks(system),
            monitoring_implemented=self.check_monitoring(system)
        )

class ManageFunction:
    async def assess(self, system: AISystem) -> ManageResult:
        return ManageResult(
            risks_prioritized=self.check_risk_prioritization(system),
            resources_allocated=self.check_resources(system),
            monitoring_continuous=self.check_continuous_monitoring(system),
            improvement_process=self.check_improvement_process(system)
        )
```

---

## Industry-Specific Compliance

### HIPAA (Healthcare)
```python
class HIPAACompliance:
    def __init__(self, encryption: EncryptionService, audit: AuditLogger):
        self.encryption = encryption
        self.audit = audit
    
    async def ensure_compliance(self, system: System) -> HIPAAReport:
        checks = await asyncio.gather(
            self.check_encryption_at_rest(),
            self.check_encryption_in_transit(),
            self.check_access_controls(),
            self.check_audit_controls(),
            self.check_integrity_controls(),
            self.check_transmission_security(),
            self.check_business_associate_agreements(),
            self.check_workforce_training(),
            self.check_incident_response(),
            self.check_contingency_plan()
        )
        
        return HIPAAReport(
            compliant=all(c.passed for c in checks),
            details=checks
        )
    
    async def check_encryption_at_rest(self) -> CheckResult:
        # Verify AES-256 encryption for PHI at rest
        return CheckResult(
            passed=await self.verify_aes_256(),
            details="All PHI encrypted with AES-256 at rest"
        )
    
    async def check_access_controls(self) -> CheckResult:
        # Unique user IDs, emergency access, automatic logoff
        return CheckResult(
            passed=await self.verify_access_controls(),
            details="Unique user IDs, automatic logoff, emergency access procedures"
        )
```

### PCI-DSS (Payment Cards)
```python
class PCIDSSCompliance:
    def __init__(self):
        self.requirements = {
            "1": "Firewall configuration",
            "2": "Default passwords changed",
            "3": "Protect stored cardholder data",
            "4": "Encrypt transmission",
            "5": "Anti-virus software",
            "6": "Secure systems/applications",
            "7": "Restrict access by need-to-know",
            "8": "Unique IDs for access",
            "9": "Physical access restriction",
            "10": "Track and monitor access",
            "11": "Regular testing",
            "12": "Information security policy"
        }
    
    async def assess(self, system: System) -> PCIReport:
        results = {}
        for req_id, description in self.requirements.items():
            check = getattr(self, f"check_req_{req_id}")()
            results[req_id] = await check
        
        return PCIReport(
            compliant=all(r.passed for r in results.values()),
            details=results,
            attestation=self.generate_attestation(results)
        )
```

### SOX (Financial Reporting)
```python
class SOXCompliance:
    async def assess_financial_controls(self, system: System) -> SOXReport:
        # Section 302: Corporate responsibility for financial reports
        # Section 404: Management assessment of internal controls
        # Section 409: Real-time issuer disclosures
        
        controls = await asyncio.gather(
            self.check_access_controls(),
            self.check_change_management(),
            self.check_segregation_of_duties(),
            self.check_audit_trail_integrity(),
            self.check_data_integrity(),
            self.check_backup_recovery()
        )
        
        return SOXReport(
            section_302=controls[0],
            section_404=controls[1],
            section_409=controls[2],
            overall=all(c.passed for c in controls)
        )
```

---

## Compliance Automation

### Continuous Compliance Monitoring
```python
class ComplianceMonitor:
    def __init__(self, frameworks: list[str]):
        self.frameworks = frameworks
        self.checks = self.load_checks(frameworks)
        self.scheduler = AsyncScheduler()
    
    async def start_monitoring(self):
        # Schedule continuous checks
        self.scheduler.every(1).hour.do(self.run_checks)
        self.scheduler.every(1).day.do(self.generate_daily_report)
        self.scheduler.every(7).days.do(self.generate_weekly_report)
    
    async def run_checks(self):
        results = await asyncio.gather(*[
            check.execute() for check in self.checks
        ])
        
        # Alert on failures
        for result in results:
            if not result.passed:
                await self.alert_on_failure(result)
    
    async def generate_evidence_package(self, framework: str) -> EvidencePackage:
        """Generate audit-ready evidence package"""
        checks = await self.run_framework_checks(framework)
        
        return EvidencePackage(
            framework=framework,
            generated_at=datetime.utcnow(),
            controls=self.map_controls_to_evidence(framework),
            test_results=self.get_test_results(),
            artifacts=self.collect_artifacts(),
            attestation=self.generate_attestation()
        )
```

### Policy as Code (OPA)
```rego
# policies/compliance.rego
package compliance

# GDPR: Personal data must be encrypted at rest
deny[msg] {
    resource := input.resource
    resource.type == "database"
    not resource.encryption_at_rest
    msg := sprintf("Database %s must have encryption at rest", [resource.name])
}

# GDPR: Personal data encrypted in transit
deny[msg] {
    resource := input.resource
    resource.type == "api_endpoint"
    not resource.tls_enabled
    msg := sprintf("API %s must use TLS", [resource.name])
}

# PCI-DSS: No default passwords
deny[msg] {
    resource := input.resource
    resource.type == "database"
    resource.default_password == true
    msg := sprintf("Database %s uses default password", [resource.name])
}

# HIPAA: Access control
deny[msg] {
    resource := input.resource
    resource.type == "phi_datastore"
    not resource.unique_user_ids
    msg := sprintf("PHI datastore %s lacks unique user IDs", [resource.name])
}

# SOX: Change management
deny[msg] {
    change := input.change
    change.type == "production_deployment"
    not change.approved_by_manager
    msg := sprintf("Production change %s not approved by manager", [change.id])
}

# Test
test_gdpr_encryption {
    input := {
        resource: {"type": "database", "name": "users_db", "encryption_at_rest": false}
    }
    deny[0] == "Database users_db must have encryption at rest"
}
```

---

## Audit & Evidence

### Audit Trail Generation
```python
class AuditTrailGenerator:
    def __init__(self, event_store: EventStore):
        self.events = event_store
    
    async def generate_audit_trail(self, 
                                   entity: str, 
                                   entity_id: str,
                                   start: datetime,
                                   end: datetime) -> AuditTrail:
        events = await self.events.query(
            entity=entity,
            entity_id=entity_id,
            start=start,
            end=end
        )
        
        return AuditTrail(
            entity=entity,
            entity_id=entity_id,
            period=(start, end),
            events=events,
            summary=self.generate_summary(events),
            integrity_hash=self.compute_integrity_hash(events)
        )
    
    def generate_integrity_hash(self, events: list[Event]) -> str:
        """Hash chain for integrity verification"""
        chain = ""
        for event in events:
            chain = hashlib.sha256(f"{chain}{event.id}{event.timestamp}".encode()).hexdigest()
        return chain
```

### Evidence Collection for Audits
```python
class EvidenceCollector:
    def __init__(self, storage: EvidenceStorage):
        self.storage = storage
    
    async def collect_evidence(self, framework: str, 
                               control_id: str) -> EvidencePackage:
        """Collect all evidence for a specific control"""
        evidence = EvidencePackage(
            control_id=control_id,
            framework=self.get_framework(control_id),
            collected_at=datetime.utcnow(),
            artifacts=[]
        )
        
        # Collect different evidence types
        evidence.artifacts.extend(await self.collect_configs())
        evidence.artifacts.extend(await self.collect_logs())
        evidence.artifacts.extend(await self.collect_test_results())
        evidence.artifacts.extend(await self.collect_procedures())
        evidence.artifacts.extend(await self.collect_training_records())
        evidence.artifacts.extend(await self.collect_incident_records())
        
        # Package and sign
        evidence.package_hash = self.compute_hash(evidence)
        evidence.signed_by = await self.sign(evidence)
        
        await self.storage.store(evidence)
        return evidence
    
    async def collect_configs(self) -> list[Artifact]:
        artifacts = []
        for config in await self.get_all_configs():
            artifacts.append(Artifact(
                type="configuration",
                name=config.name,
                content=config.content,
                hash=self.hash(config.content),
                collected_at=datetime.utcnow()
            ))
        return artifacts
```

---

## Vendor Risk Management

### Vendor Assessment
```python
class VendorRiskManager:
    def __init__(self):
        self.vendors: dict[str, VendorProfile] = {}
        self.questionnaires = {
            "security": SecurityQuestionnaire(),
            "privacy": PrivacyQuestionnaire(),
            "business_continuity": BCQuestionnaire(),
            "financial": FinancialQuestionnaire()
        }
    
    async def assess_vendor(self, vendor: Vendor) -> VendorRiskAssessment:
        # Send questionnaires
        responses = await asyncio.gather(*[
            q.send(vendor.contact) for q in self.questionnaires.values()
        ])
        
        # Analyze responses
        scores = {name: q.score(r) for (name, q), r in zip(self.questionnaires.items(), responses)}
        
        # Security certifications
        certs = await self.verify_certifications(vendor)
        
        # Subprocessor review
        subprocessors = await self.review_subprocessors(vendor)
        
        return VendorRiskAssessment(
            vendor=vendor,
            overall_score=statistics.mean(scores.values()),
            scores=scores,
            certifications=certs,
            subprocessors=subprocessors,
            risk_level=self.calculate_risk_level(scores),
            recommendations=self.generate_recommendations(scores)
        )
    
    async def continuous_monitoring(self, vendor_id: str):
        """Continuous monitoring of vendor risk"""
        while True:
            # Check for security incidents
            incidents = await self.check_security_incidents(vendor_id)
            
            # Check financial health
            financial = await self.check_financial_health(vendor_id)
            
            # Check certification status
            certs = await self.verify_certifications(vendor_id)
            
            if incidents or financial.declining or certs.expiring:
                await self.escalate(vendor_id, incidents, financial, certs)
            
            await asyncio.sleep(86400)  # Daily
```

---

## Training & Awareness

### Compliance Training Program
```python
class ComplianceTrainingProgram:
    def __init__(self, lms: LMSClient):
        self.lms = lms
        self.courses = {
            "gdpr_basics": Course("GDPR Basics", duration="2h", mandatory=True),
            "hipaa_compliance": Course("HIPAA Compliance", duration="3h", mandatory=True),
            "pci_dss": Course("PCI-DSS Basics", duration="2h", mandatory="role:payment"),
            "secure_coding": Course("Secure Coding Practices", duration="4h", mandatory="role:developer"),
            "incident_response": Course("Incident Response", duration="2h", mandatory="role:engineer"),
            "ai_ethics": Course("AI Ethics & Safety", duration="2h", mandatory="role:ai_engineer"),
            "data_handling": Course("Data Handling & Privacy", duration="1h", mandatory=True)
        }
    
    async def assign_training(self, user: User, role: str):
        required = [c for c in self.courses.values() 
                   if c.mandatory == True or c.mandatory == f"role:{role}"]
        
        for course in required:
            await self.lms.assign(user.id, course.id, deadline=timedelta(days=30))
    
    async def track_completion(self, user_id: str) -> TrainingStatus:
        completions = await self.lms.get_completions(user_id)
        required = self.get_required_courses(user_id)
        
        return TrainingStatus(
            completed=[c for c in required if c.id in completions],
            pending=[c for c in required if c.id not in completions],
            overdue=[c for c in required if c.id not in completions and c.deadline < datetime.utcnow()],
            compliance_rate=len([c for c in required if c.id in completions]) / len(required)
        )
```

---

## Related Notes

- [[Production/Security.md]] — Security controls
- [[Production/Observability.md]] — Audit logging
- [[Production/Reliability.md]] — Incident response
- [[Production/Deployment-Architecture.md]] — Deployment compliance
- [[Production/Cost-Optimization.md]] — Cost compliance
- [[Security/Secure-Coding.md]] — Secure coding
- [[Security/Supply-Chain-Security.md]] — Supply chain compliance
- [[Production/Deployment-Architecture.md]] — Deployment compliance
- [[Production/Reliability.md]] — Incident response
- [[Production/Observability.md]] — Audit logging
- [[Security/Secure-Coding.md]] — Secure coding
- [[Security/Supply-Chain-Security.md]] — Supply chain compliance
- [[DevOps/CI-CD-Best-Practices.md]] — CI/CD compliance
- [[DevOps/Infrastructure-as-Code.md]] — IaC compliance
- [[Frontiers/Research-Directions.md]] — Compliance research
- [[Frontiers/Agent-Societies.md]] — AI governance
- [[Security/Secure-Coding.md]] — Secure coding
- [[Security/Supply-Chain-Security.md]] — Supply chain compliance

---

*Sources: GDPR, CCPA/CPRA, HIPAA, PCI-DSS, SOX, EU AI Act, NIST AI RMF, ISO 27001, SOC 2, FedRAMP, StateRAMP, CMMC, ISO 42001, NIST AI RMF, ISO 22989, ISO 23894, ISO 42001, OWASP LLM Top 10, NIST AI RMF Playbook, MITRE ATLAS, OWASP LLM AI Security, FAIR Model, FAIR Institute, Factor Analysis of Information Risk, FAIR-U, OpenFAIR, RiskLens, RiskOptics, Vanta, Drata, Secureframe, Tugboat Logic, Hyperproof, AuditBoard, Workiva, AuditBoard, LogicGate, Riskonnect, MetricStream, RSA Archer, ServiceNow GRC, MetricStream, NAVEX Global, OneTrust, TrustArc, DataGrail, Ketch, Osano, Ethyca, Transcend, Mine, BigID, OneTrust, TrustArc, DataGrail.*