The Air-Gapped Chronicles: The Insurance Gap — Building Liability-Resistant AI When Insurance Won't Cover the Risk

from fastapi import FastAPI, HTTPException, BackgroundTasks
from pydantic import BaseModel
from typing import Optional, Literal
import redis
import json
import hashlib
from datetime import datetime, timedelta
import asyncio

app = FastAPI()
redis_client = redis.Redis(host='localhost', port=6379, decode_responses=True)

class AIRecommendation(BaseModel):
    recommendation_id: str
    recommendation_type: Literal['medical_diagnosis', 'financial_approval', 'legal_advice']
    ai_output: str
    risk_level: Literal['low', 'medium', 'high', 'critical']
    context: dict
    requires_approval: bool
    generated_at: str

class ApprovalDecision(BaseModel):
    recommendation_id: str
    decision: Literal['approved', 'rejected']
    approver_id: str
    reason: Optional[str] = None

class HumanInLoopSystem:
    APPROVAL_QUEUE = "approval_queue"
    APPROVED_SET = "approved_recommendations"
    REJECTED_SET = "rejected_recommendations"
    APPROVAL_TIMEOUT_HOURS = 24

    def __init__(self):
        self.redis = redis_client

    async def submit_for_approval(self, recommendation: AIRecommendation) -> dict:
        if not self._requires_approval(recommendation):
            return {
                'status': 'auto_approved',
                'recommendation_id': recommendation.recommendation_id,
                'approved_at': datetime.utcnow().isoformat()
            }

        queue_data = {
            'recommendation': recommendation.dict(),
            'submitted_at': datetime.utcnow().isoformat(),
            'expires_at': (datetime.utcnow() + timedelta(hours=self.APPROVAL_TIMEOUT_HOURS)).isoformat()
        }

        self.redis.lpush(self.APPROVAL_QUEUE, json.dumps(queue_data))
        queue_length = self.redis.llen(self.APPROVAL_QUEUE)

        return {
            'status': 'pending_approval',
            'recommendation_id': recommendation.recommendation_id,
            'queue_position': queue_length,
            'estimated_wait_minutes': queue_length * 5,
            'expires_at': queue_data['expires_at']
        }

    def _requires_approval(self, recommendation: AIRecommendation) -> bool:
        if recommendation.risk_level in ['high', 'critical']:
            return True
        if recommendation.recommendation_type in ['medical_diagnosis', 'legal_advice']:
            return True
        if recommendation.recommendation_type == 'financial_approval':
            if recommendation.context.get('amount', 0) > 10000:
                return True
        return False

import re
from typing import List, Dict, Optional
import anthropic
from presidio_analyzer import AnalyzerEngine
from presidio_anonymizer import AnonymizerEngine

class OutputValidator:
    def __init__(self):
        self.analyzer = AnalyzerEngine()
        self.anonymizer = AnonymizerEngine()

        self.harmful_patterns = [
            r'\b(kill yourself|end it all|you should die)\b',
            r'\b(methods of suicide|how to commit suicide)\b',
            r'\b(build a bomb|make explosives|hurt someone)\b',
            r'\b(how to hack|steal credit card|forge document)\b'
        ]

    async def validate(self, output: str, context: dict = None) -> Dict:
        violations = []
        risk_score = 0.0
        sanitized = output

        # Check 1: Harmful content
        harmful_check = self._check_harmful_content(output)
        if harmful_check['detected']:
            violations.append(f"Harmful content: {harmful_check['type']}")
            risk_score += 0.8

        # Check 2: PII/PHI leakage
        pii_check = self._check_pii_leakage(output)
        if pii_check['detected']:
            violations.append(f"PII detected: {', '.join(pii_check['types'])}")
            sanitized = pii_check['sanitized']
            risk_score += 0.6

        # Check 3: Hallucinated citations
        citation_check = self._check_citations(output)
        if citation_check['suspicious']:
            violations.append(f"Suspicious citations: {citation_check['count']}")
            risk_score += 0.4

        valid = risk_score < 0.5 and len(violations) == 0

        return {
            'valid': valid,
            'violations': violations,
            'sanitized_output': sanitized if valid else None,
            'risk_score': risk_score
        }

import hashlib
import json
from datetime import datetime
from typing import Dict, List, Optional
import psycopg2

class CryptographicAuditLog:
    def __init__(self, db_connection_string: str):
        self.conn = psycopg2.connect(db_connection_string)

    def log_event(self, event_type, user_id, ai_model, 
                  input_data, output_data, decision, metadata=None) -> str:
        timestamp = datetime.utcnow()
        input_hash = self._hash_data(input_data)
        output_hash = self._hash_data(output_data)
        previous_hash = self._get_last_hash()

        log_entry = {
            'timestamp': timestamp.isoformat(),
            'event_type': event_type,
            'user_id': user_id,
            'ai_model': ai_model,
            'input_hash': input_hash,
            'output_hash': output_hash,
            'decision': decision,
            'metadata': metadata or {},
            'previous_hash': previous_hash
        }

        current_hash = self._hash_data(json.dumps(log_entry, sort_keys=True))
        # Store in database with hash chain
        return current_hash

    def verify_chain_integrity(self) -> Dict:
        """Verify entire audit log chain is intact"""
        # Each entry's previous_hash must match the prior entry's current_hash
        # Any tampering breaks the chain and is detectable
        pass

import redis
from datetime import datetime, timedelta
from enum import Enum

class SystemStatus(str, Enum):
    HEALTHY = "healthy"
    DEGRADED = "degraded"
    EMERGENCY_SHUTDOWN = "emergency_shutdown"

class CircuitBreakerState(str, Enum):
    CLOSED = "closed"      # System operational
    OPEN = "open"          # System shut down
    HALF_OPEN = "half_open" # Testing recovery

class AIKillSwitch:
    FAILURE_THRESHOLD = 10
    SUCCESS_THRESHOLD = 5
    TIMEOUT_SECONDS = 300

    def emergency_shutdown(self, authorized_by: str, reason: str):
        """Immediate shutdown - requires authorization"""
        self.redis.set('ai_system_status', SystemStatus.EMERGENCY_SHUTDOWN)
        self.redis.set('circuit_breaker_state', CircuitBreakerState.OPEN)
        self._send_alert(severity='critical', 
                        message=f'EMERGENCY SHUTDOWN by {authorized_by}: {reason}')