Financial AI Agent Guardrails

import { createHash } from 'crypto';

interface AuditEntry {
  timestamp: string;
  agentId: string;
  action: string;
  inputHash: string;
  outputHash: string;
  decision: string;
  userId: string;
  previousHash: string;
  signature: string;
}

class SOXAuditLogger {
  private chain: AuditEntry[] = [];

  async logDecision(
    agentId: string,
    action: string,
    input: unknown,
    output: unknown,
    userId: string
  ): Promise<AuditEntry> {
    const entry: AuditEntry = {
      timestamp: new Date().toISOString(),
      agentId,
      action,
      inputHash: this.hash(input),
      outputHash: this.hash(output),
      decision: JSON.stringify(output),
      userId,
      previousHash: this.chain.length > 0
        ? this.chain[this.chain.length - 1].signature
        : 'genesis',
      signature: ''
    };

    // Create cryptographic signature for tamper-proof record
    entry.signature = this.signEntry(entry);
    this.chain.push(entry);

    // Persist to immutable storage (e.g., WORM-compatible)
    await this.persistToImmutableStorage(entry);

    return entry;
  }

  private hash(data: unknown): string {
    return createHash('sha256')
      .update(JSON.stringify(data))
      .digest('hex');
  }

  private signEntry(entry: Omit<AuditEntry, 'signature'>): string {
    return createHash('sha256')
      .update(JSON.stringify(entry))
      .digest('hex');
  }

  async persistToImmutableStorage(entry: AuditEntry): Promise<void> {
    // Implementation for SEC 17a-4 compliant storage
    // Use AWS S3 Object Lock, Azure Immutable Storage, etc.
  }

  verifyIntegrity(): boolean {
    for (let i = 1; i < this.chain.length; i++) {
      if (this.chain[i].previousHash !== this.chain[i-1].signature) {
        return false;
      }
    }
    return true;
  }
}

interface PaymentData {
  token: string;
  lastFour: string;
  cardType: string;
  expiryMonth: string;
  expiryYear: string;
}

class PCICompliantPaymentHandler {
  private encryptionKey: Buffer;
  private allowedActions: Set<string>;

  constructor() {
    // PCI DSS Requirement 3: Protect stored cardholder data
    this.encryptionKey = this.loadEncryptionKey();
    this.allowedActions = new Set(['authorize', 'capture', 'refund', 'void']);
  }

  // Never store or log PAN - PCI DSS Requirement 3.2
  async tokenizeCardData(pan: string): Promise<PaymentData> {
    // PCI DSS Requirement 3.4: Render PAN unreadable via tokenization
    const token = await this.tokenizationService.tokenize(pan);

    // PCI DSS Requirement 3.3: Mask PAN when displayed
    return {
      token,
      lastFour: pan.slice(-4),
      cardType: this.detectCardType(pan),
      expiryMonth: '',
      expiryYear: ''
    };
  }

  // PCI DSS Requirement 7: Restrict access by business need-to-know
  async processPayment(
    agentId: string,
    action: string,
    token: string,
    amount: number,
    userId: string
  ): Promise<{ success: boolean; transactionId: string }> {

    // Verify action is allowed for this agent
    if (!this.allowedActions.has(action)) {
      throw new Error(`Action '${action}' not permitted for agent ${agentId}`);
    }

    // PCI DSS Requirement 8: Identify users and authenticate access
    const isAuthorized = await this.verifyUserPermission(userId, 'payment:process');
    if (!isAuthorized) {
      throw new Error('User not authorized for payment processing');
    }

    // PCI DSS Requirement 10: Track and monitor all access
    await this.auditLog.log({
      timestamp: new Date().toISOString(),
      agentId,
      action,
      token, // Token, not PAN
      amount,
      userId,
      result: 'processing'
    });

    return await this.paymentGateway.process(token, amount);
  }
}

from dataclasses import dataclass
from datetime import datetime
from typing import Optional, Dict, Any, List
from enum import Enum

class ModelRiskTier(Enum):
    TIER_1_HIGH = "tier_1_high"      # Credit decisions, trading
    TIER_2_MEDIUM = "tier_2_medium"   # Fraud detection, marketing
    TIER_3_LOW = "tier_3_low"        # Internal operations

@dataclass
class ModelDocumentation:
    model_id: str
    model_name: str
    version: str
    developer: str
    business_owner: str
    risk_tier: ModelRiskTier
    purpose: str
    methodology: str
    limitations: List[str]
    input_features: List[str]
    validation_status: str
    last_validation_date: datetime
    next_validation_date: datetime

class FFIECModelGovernance:
    def __init__(self, model_registry, validation_service):
        self.registry = model_registry
        self.validator = validation_service

    def register_model(self, model, documentation):
        """SR 11-7: Maintain comprehensive model inventory"""
        model_id = self.registry.register(model, documentation)

        # Assign validation frequency based on risk tier
        validation_frequency = {
            ModelRiskTier.TIER_1_HIGH: 90,   # Quarterly
            ModelRiskTier.TIER_2_MEDIUM: 180, # Semi-annual
            ModelRiskTier.TIER_3_LOW: 365     # Annual
        }

        return model_id

    def validate_model(self, model_id):
        """SR 11-7: Independent model validation"""
        model_doc = self.registry.get_documentation(model_id)
        model = self.registry.get_model(model_id)

        validation_result = self.validator.validate(
            model=model,
            documentation=model_doc,
            tests=[
                'conceptual_soundness',  # Theory and methodology
                'data_quality',          # Input data assessment
                'developmental_evidence', # Back-testing
                'outcome_analysis',      # Actual vs predicted
                'sensitivity_analysis',  # Stress testing
                'bias_fairness'          # For credit models
            ]
        )

        return validation_result

    def can_deploy_model(self, model_id, use_case):
        """SR 11-7: Deployment approval process"""
        doc = self.registry.get_documentation(model_id)
        validation = self.registry.get_latest_validation(model_id)

        checks = [
            validation.status == 'approved',
            doc.next_validation_date > datetime.now(),
            use_case in doc.purpose,
            self.is_risk_committee_approved(model_id)
        ]

        return all(checks)

policies:
  # Transaction limit enforcement
  - name: wire_transfer_limit
    tool: wire_transfer
    rules:
      - condition: "amount > 100000"
        action: require_approval
        approvers: ["cfo@company.com", "treasury@company.com"]
      - condition: "amount > 50000 && amount <= 100000"
        action: require_approval
        approvers: ["treasury@company.com"]
      - condition: "destination_account.new_vendor == true"
        action: require_approval
        approvers: ["accounting@company.com"]

  # Invoice validation
  - name: invoice_processing
    tool: process_invoice
    rules:
      - condition: "vendor.approved == false"
        action: deny
        reason: "Vendor not in approved vendor master"
      - condition: "vendor_bank_account.new == true"
        action: require_approval
        approvers: ["ap-manager@company.com"]

  # Refund controls
  - name: refund_processing
    tool: process_refund
    rules:
      - condition: "refund_amount > original_amount"
        action: deny
        reason: "Refund cannot exceed original transaction"
      - condition: "refund_amount / original_amount > 1.1"
        action: require_approval

  # PII protection
  - name: account_data_access
    tool: get_account_details
    rules:
      - condition: "requesting_user != account_owner"
        action: deny
        reason: "Cross-customer data access blocked"
      - condition: "fields_requested includes 'full_account_number'"
        action: deny
        reason: "Full account numbers cannot be returned"