Healthcare AI Agent Guardrails

import { AuditLogger } from './audit-logger';
import { PHIService } from './phi-service';

interface AgentContext {
  agentId: string;
  role: 'clinical_assistant' | 'admin_bot' | 'scheduling_agent';
  department: string;
  purpose: 'treatment' | 'payment' | 'operations' | 'research';
  sessionId: string;
}

interface AccessDecision {
  allowed: boolean;
  scope: string[];
  reason: string;
  conditions?: string[];
}

class HIPAAAccessController {
  private auditLogger: AuditLogger;
  private phiService: PHIService;

  // Minimum necessary scope definitions by role and purpose
  private readonly minimumNecessaryScopes: Record<string, string[]> = {
    'clinical_assistant:treatment': ['Patient.read', 'Observation.read', 'Condition.read'],
    'scheduling_agent:operations': ['Patient.read.name', 'Patient.read.contact', 'Appointment.*'],
    'admin_bot:payment': ['Patient.read.demographics', 'Claim.*', 'Coverage.read'],
  };

  // Sensitive data categories requiring additional consent
  private readonly sensitiveCategories = new Set([
    'psychotherapy_notes',
    'hiv_status',
    'substance_abuse',
    'reproductive_health',
    'genetic_information'
  ]);

  async evaluateAccess(
    context: AgentContext,
    requestedResource: string,
    requestedAction: 'read' | 'write' | 'delete'
  ): Promise<AccessDecision> {
    const decision: AccessDecision = {
      allowed: false,
      scope: [],
      reason: '',
      conditions: []
    };

    // Build access key and retrieve minimum necessary scope
    const accessKey = `${context.role}:${context.purpose}`;
    const permittedScopes = this.minimumNecessaryScopes[accessKey] || [];

    // Check if requested resource falls within minimum necessary
    const isWithinScope = permittedScopes.some(scope =>
      this.matchesScope(requestedResource, requestedAction, scope)
    );

    if (!isWithinScope) {
      decision.reason = `Resource ${requestedResource} exceeds minimum necessary`;
      await this.auditLogger.logDeniedAccess({
        agentId: context.agentId,
        resource: requestedResource,
        action: requestedAction,
        reason: decision.reason,
        sessionId: context.sessionId,
        timestamp: new Date().toISOString()
      });
      return decision;
    }

    // Check for sensitive data segmentation
    if (this.isSensitiveCategory(requestedResource)) {
      const hasConsent = await this.checkSegmentationConsent(requestedResource, context);
      if (!hasConsent) {
        decision.reason = 'Sensitive data category requires additional patient consent';
        decision.conditions = ['patient_consent_required'];
        await this.auditLogger.logDeniedAccess({
          agentId: context.agentId,
          resource: requestedResource,
          action: requestedAction,
          reason: decision.reason,
          sessionId: context.sessionId,
          timestamp: new Date().toISOString()
        });
        return decision;
      }
    }

    decision.allowed = true;
    decision.scope = permittedScopes;
    decision.reason = 'Access granted within minimum necessary standard';

    // Log approved access
    await this.auditLogger.logAccess({
      agentId: context.agentId,
      resource: requestedResource,
      action: requestedAction,
      scope: permittedScopes,
      sessionId: context.sessionId,
      timestamp: new Date().toISOString(),
      purpose: context.purpose
    });

    return decision;
  }

  private matchesScope(resource: string, action: string, scope: string): boolean {
    const [scopeResource, scopeAction] = scope.split('.');
    const actionMatch = scopeAction === '*' || scopeAction === action;
    const resourceMatch = resource.startsWith(scopeResource) || scopeResource === '*';
    return actionMatch && resourceMatch;
  }

  private isSensitiveCategory(resource: string): boolean {
    return Array.from(this.sensitiveCategories).some(cat =>
      resource.toLowerCase().includes(cat)
    );
  }

  private async checkSegmentationConsent(resource: string, context: AgentContext): Promise<boolean> {
    return this.phiService.verifyConsent(resource, context.purpose);
  }
}

import crypto from 'crypto';

interface AuditEntry {
  id: string;
  timestamp: string;
  agentId: string;
  sessionId: string;
  action: 'PHI_ACCESS' | 'PHI_READ' | 'PHI_WRITE' | 'AI_INFERENCE' | 'DENIED_ACCESS';
  resource: string;
  dataType: string;
  patientId?: string;
  outcome: 'SUCCESS' | 'FAILURE' | 'ERROR';
  details: Record<string, unknown>;
  ipAddress: string;
  userAgent: string;
  dataHash?: string;
  previousHash: string; // Blockchain-style chaining for tamper detection
}

class HIPAAAuditLogger {
  private readonly logStore: AuditLogStore;
  private lastHash: string = '0'.repeat(64); // Genesis hash

  async logAccess(entry: Omit<AuditEntry, 'id' | 'previousHash'>): Promise<void> {
    const auditEntry: AuditEntry = {
      ...entry,
      id: crypto.randomUUID(),
      previousHash: this.lastHash
    };

    // Create hash chain for integrity
    const entryHash = this.createEntryHash(auditEntry);
    this.lastHash = entryHash;

    // Sign entry for authenticity
    const signature = await this.signEntry(auditEntry);

    await this.logStore.append({
      ...auditEntry,
      signature
    });
  }

  async logAIInference(params: {
    agentId: string;
    sessionId: string;
    modelVersion: string;
    inputTokens: number;
    outputTokens: number;
    promptHash: string;
    responseHash: string;
    phiAccessed: string[];
    decision: string;
    confidence?: number;
    humanReviewRequired: boolean;
  }): Promise<void> {
    await this.logAccess({
      timestamp: new Date().toISOString(),
      agentId: params.agentId,
      sessionId: params.sessionId,
      action: 'AI_INFERENCE',
      resource: `model:${params.modelVersion}`,
      dataType: 'AI_MODEL_OUTPUT',
      outcome: 'SUCCESS',
      details: {
        inputTokens: params.inputTokens,
        outputTokens: params.outputTokens,
        promptHash: params.promptHash,
        responseHash: params.responseHash,
        phiAccessed: params.phiAccessed,
        decision: params.decision,
        confidence: params.confidence,
        humanReviewRequired: params.humanReviewRequired
      },
      ipAddress: 'system',
      userAgent: `AI-Agent/${params.agentId}`
    });
  }

  private createEntryHash(entry: AuditEntry): string {
    const entryString = JSON.stringify({
      timestamp: entry.timestamp,
      agentId: entry.agentId,
      action: entry.action,
      resource: entry.resource,
      outcome: entry.outcome,
      previousHash: entry.previousHash
    });
    return crypto.createHash('sha256').update(entryString).digest('hex');
  }

  // Verify log integrity - detect tampering
  async verifyIntegrity(): Promise<{ valid: boolean; errors: string[] }> {
    const entries = await this.logStore.getAll();
    const errors: string[] = [];

    for (let i = 0; i < entries.length; i++) {
      const expectedPrevHash = i === 0 ? '0'.repeat(64) : entries[i - 1].hash;

      if (entries[i].previousHash !== expectedPrevHash) {
        errors.push(`Chain break at entry ${entries[i].id}`);
      }

      const isValid = await this.verifySignature(entries[i]);
      if (!isValid) {
        errors.push(`Invalid signature at entry ${entries[i].id}`);
      }
    }

    return { valid: errors.length === 0, errors };
  }
}

import re
from typing import List, Dict, Any
from dataclasses import dataclass
from enum import Enum

class PHIType(Enum):
    NAME = "name"
    SSN = "ssn"
    MRN = "mrn"
    DATE = "date"
    PHONE = "phone"
    EMAIL = "email"
    ADDRESS = "address"

@dataclass
class PHIDetectionResult:
    contains_phi: bool
    phi_types: List[PHIType]
    redacted_text: str
    phi_count: int

class PHISanitizer:
    """HIPAA-compliant PHI sanitization for AI agent inputs/outputs."""

    # HIPAA Safe Harbor identifiers regex patterns
    PHI_PATTERNS = {
        PHIType.SSN: re.compile(r'\b\d{3}-\d{2}-\d{4}\b|\b\d{9}\b'),
        PHIType.MRN: re.compile(r'\bMRN[:\s]*(\d{6,12})\b', re.IGNORECASE),
        PHIType.PHONE: re.compile(
            r'\b(?:\+?1[-.]?)?\(?\d{3}\)?[-.]?\d{3}[-.]?\d{4}\b'
        ),
        PHIType.EMAIL: re.compile(
            r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b'
        ),
        PHIType.DATE: re.compile(
            r'\b\d{1,2}[/-]\d{1,2}[/-]\d{2,4}\b|'
            r'\b(?:Jan|Feb|Mar|Apr|May|Jun|Jul|Aug|Sep|Oct|Nov|Dec)[a-z]* '
            r'\d{1,2},? \d{4}\b',
            re.IGNORECASE
        ),
    }

    def __init__(self, minimum_necessary_fields: Dict[str, List[str]]):
        self.minimum_necessary_fields = minimum_necessary_fields

    def detect_and_redact(
        self,
        text: str,
        allowed_phi_types: List[PHIType]
    ) -> PHIDetectionResult:
        """Detect PHI in text and redact types not in allowed list."""
        phi_types_found = []
        redacted_text = text
        phi_count = 0

        for phi_type, pattern in self.PHI_PATTERNS.items():
            matches = pattern.findall(text)
            if matches:
                phi_types_found.append(phi_type)
                phi_count += len(matches)

                if phi_type not in allowed_phi_types:
                    # Redact PHI not allowed by minimum necessary
                    redacted_text = pattern.sub(
                        f'[{phi_type.value.upper()}_REDACTED]',
                        redacted_text
                    )

        return PHIDetectionResult(
            contains_phi=len(phi_types_found) > 0,
            phi_types=phi_types_found,
            redacted_text=redacted_text,
            phi_count=phi_count
        )


class MinimumNecessaryEnforcer:
    """Enforces HIPAA minimum necessary standard for AI data access."""

    def __init__(self):
        self.scope_definitions = {
            'clinical_assistant:treatment': {
                'patient_fields': ['name', 'dob', 'diagnosis', 'medications', 'allergies'],
                'exclude_fields': ['ssn', 'financial_info', 'insurance_account'],
                'date_precision': 'month_only'  # Reduce date specificity
            },
            'scheduling_agent:operations': {
                'patient_fields': ['name', 'phone', 'preferred_times'],
                'exclude_fields': ['diagnosis', 'medications', 'ssn', 'mrn'],
                'date_precision': 'full'
            },
            'billing_agent:payment': {
                'patient_fields': ['name', 'dob', 'insurance_info', 'account_number'],
                'exclude_fields': ['diagnosis', 'medications', 'clinical_notes'],
                'date_precision': 'full'
            }
        }

    def filter_data(
        self,
        data: Dict[str, Any],
        role: str,
        purpose: str
    ) -> Dict[str, Any]:
        """Filter data to minimum necessary for role/purpose."""
        scope_key = f"{role}:{purpose}"
        scope = self.scope_definitions.get(scope_key)

        if not scope:
            # Deny all if scope not defined - fail secure
            return {'error': 'No access scope defined', 'data': None}

        filtered = {}
        allowed_fields = scope['patient_fields']
        excluded = scope['exclude_fields']
        date_precision = scope['date_precision']

        for key, value in data.items():
            if key in excluded:
                continue

            if key in allowed_fields:
                # Apply date precision if applicable
                if 'date' in key.lower() or 'dob' in key.lower():
                    value = self._apply_date_precision(value, date_precision)

                filtered[key] = value

        return filtered

    def _apply_date_precision(self, date_value: str, precision: str) -> str:
        """Reduce date precision to minimum necessary."""
        if precision == 'month_only':
            return re.sub(r'-(\d{2})$', '-XX', date_value)
        elif precision == 'year_only':
            return date_value[:4] + '-XX-XX'
        return date_value


# Usage example
sanitizer = PHISanitizer(
    minimum_necessary_fields={
        'scheduling_agent:operations': ['name', 'phone'],
        'clinical_assistant:treatment': ['name', 'mrn', 'date']
    }
)

# Sanitize a scheduling prompt
prompt = """Schedule appointment for John Doe, SSN 123-45-6789,
phone 555-123-4567, DOB 01/15/1980"""

allowed = sanitizer.get_allowed_phi_types('scheduling_agent', 'operations')
result = sanitizer.detect_and_redact(prompt, allowed)
# Output: Schedule appointment for John Doe, [SSN_REDACTED],
#         phone 555-123-4567, [DATE_REDACTED]

name: healthcare-phi-protection
description: PHI access control for clinical AI agents

rules:
  # EHR read access - minimum necessary
  - name: ehr-read-minimum-necessary
    tools: ["ehr_read", "patient_lookup"]
    action: allow
    constraints:
      fields: ["demographics", "allergies", "current_medications"]
      requires_patient_id: true
    audit:
      log_arguments: true
      retention_days: 2555  # 7 years for HIPAA

  # Block clinical notes access for non-clinical agents
  - name: block-clinical-notes
    tools: ["ehr_read"]
    arguments:
      fields: ["clinical_notes", "psychiatric_notes", "hiv_status"]
    action: deny
    response:
      error: "Access to clinical notes requires clinical staff authorization"

  # Sensitive data segmentation
  - name: sensitive-category-consent
    tools: ["ehr_read", "patient_lookup"]
    arguments:
      fields: ["psychotherapy_notes", "substance_abuse", "genetic_information"]
    action: require_approval
    constraints:
      patient_consent_verified: true
    response:
      message: "Sensitive data category requires verified patient consent"

  # Prescription limits for controlled substances
  - name: controlled-substance-check
    tools: ["prescribe_medication"]
    action: require_approval
    constraints:
      schedule: ["II", "III", "IV"]
    audit:
      log_arguments: true
      alert_on_deny: true
    response:
      approver_role: "pharmacist"
      message: "Controlled substance prescription requires pharmacist review"

  # Patient communication guardrails
  - name: patient-communication-phi-check
    tools: ["send_patient_message"]
    action: allow
    constraints:
      no_phi_in_message: true
      verified_patient_identity: true
    response:
      require_confirmation: "Confirm message contains no PHI and patient identity verified"

  # Research data - de-identification enforcement
  - name: research-data-deidentified
    tools: ["query_research_dataset"]
    action: allow
    constraints:
      dataset_type: "de_identified"
      approved_irb_protocol: true
    audit:
      log_query: true
      researcher_id: required