AI Audit Trail Requirements

/**
 * AI Decision Audit Log Schema
 *
 * Every AI inference that affects users or business outcomes
 * must produce a record conforming to this schema.
 */

interface AIDecisionLog {
  // Identity
  traceId: string;          // Distributed trace ID for request correlation
  decisionId: string;       // Unique identifier for this decision
  timestamp: string;        // ISO 8601 with timezone

  // Model context
  modelId: string;          // Model identifier (e.g., "fraud-detector")
  modelVersion: string;     // Semantic version of the model
  modelChecksum: string;    // SHA-256 of model artifact
  riskTier: "low" | "medium" | "high" | "critical";

  // Input context
  inputHash: string;        // SHA-256 of raw input (for privacy)
  inputFeatures: Record<string, number | string | boolean>;
  featureStoreVersion?: string;  // Feature store snapshot version
  contextData?: Record<string, string>;  // Additional context

  // Output
  prediction: string | number | Record<string, unknown>;
  confidence: number;       // 0.0 - 1.0
  alternativePredictions?: Array<{
    prediction: string | number;
    confidence: number;
  }>;
  explanationRef?: string;  // Reference to detailed SHAP/LIME explanation

  // Human oversight
  humanReviewRequired: boolean;
  humanReviewed: boolean;
  humanOverride: boolean;
  humanReviewerId?: string;
  humanDecision?: string;
  overrideReason?: string;

  // Impact
  affectedUserId?: string;   // Hashed user identifier
  decisionCategory: string;  // "recommendation", "scoring", "classification", etc.
  consequenceLevel: "informational" | "material" | "significant";

  // Metadata
  latencyMs: number;
  tokenUsage?: { input: number; output: number };
  environment: "production" | "staging" | "shadow";
}

/**
 * Model Lifecycle Event Log
 * Tracks training, evaluation, deployment, and retirement events.
 */
interface ModelLifecycleEvent {
  eventId: string;
  eventType:
    | "training_started"
    | "training_completed"
    | "evaluation_completed"
    | "deployment_approved"
    | "deployed"
    | "rollback"
    | "deprecated"
    | "retired";
  modelId: string;
  modelVersion: string;
  timestamp: string;
  actor: string;           // Who triggered the event
  details: Record<string, unknown>;
  trainingDataRef?: string; // Reference to training data lineage
  evaluationResults?: {
    metric: string;
    value: number;
    threshold: number;
    passed: boolean;
  }[];
}

"""Audit trail logger for AI decisions.

Provides a structured logging interface that captures
all required fields for regulatory compliance and
internal governance. Designed for high-throughput
inference pipelines.
"""

import json
import hashlib
import time
from typing import Dict, Any, Optional, List
from dataclasses import dataclass, asdict


@dataclass
class AuditEntry:
    """A single audit trail entry."""
    decision_id: str
    trace_id: str
    model_id: str
    model_version: str
    input_hash: str
    prediction: Any
    confidence: float
    risk_tier: str
    human_review_required: bool
    timestamp: float
    latency_ms: float
    environment: str


class AIAuditLogger:
    """Structured audit logger for AI decisions.

    Writes to a structured log sink (stdout as JSON by default).
    In production, configure to write to your log aggregation
    system (e.g., Elasticsearch, ClickHouse, BigQuery).
    """

    def __init__(
        self,
        model_id: str,
        model_version: str,
        risk_tier: str = "medium",
        environment: str = "production",
    ):
        self.model_id = model_id
        self.model_version = model_version
        self.risk_tier = risk_tier
        self.environment = environment

    def log_decision(
        self,
        trace_id: str,
        input_data: Dict[str, Any],
        prediction: Any,
        confidence: float,
        latency_ms: float,
        human_review_required: bool = False,
        explanation_ref: Optional[str] = None,
    ) -> AuditEntry:
        """Log an AI decision to the audit trail."""
        # Hash input to avoid logging raw PII
        input_hash = hashlib.sha256(
            json.dumps(input_data, sort_keys=True).encode()
        ).hexdigest()

        entry = AuditEntry(
            decision_id=hashlib.sha256(
                f"{trace_id}:{time.time_ns()}".encode()
            ).hexdigest()[:24],
            trace_id=trace_id,
            model_id=self.model_id,
            model_version=self.model_version,
            input_hash=input_hash,
            prediction=prediction,
            confidence=confidence,
            risk_tier=self.risk_tier,
            human_review_required=human_review_required,
            timestamp=time.time(),
            latency_ms=latency_ms,
            environment=self.environment,
        )

        # Emit structured log entry
        log_record = asdict(entry)
        if explanation_ref:
            log_record["explanation_ref"] = explanation_ref
        print(json.dumps(log_record))

        return entry