Compliance Archiving for LLM Applications: SOC2, GDPR, and HIPAA-Ready Chat Log Storage

-- PostgreSQL schema for llm_interaction_audit
CREATE TABLE llm_interaction_audit (
    -- Core Audit Trail
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    event_timestamp TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    tenant_id VARCHAR(255) NOT NULL, -- For multi-tenant SaaS cost & compliance isolation
    user_id VARCHAR(255), -- Can be NULL for system tasks, but usually required
    session_id VARCHAR(255),

    -- LLM Request Context
    provider VARCHAR(50) NOT NULL, -- e.g., 'openai', 'anthropic', 'azure', 'ollama-local'
    model VARCHAR(100) NOT NULL, -- e.g., 'gpt-4o', 'claude-3-5-sonnet'
    prompt_hash BYTEA NOT NULL, -- Cryptographic hash of the raw prompt (more on this later)
    prompt_tokens INTEGER NOT NULL,
    response_tokens INTEGER NOT NULL,
    total_cost DECIMAL(10, 6) NOT NULL, -- Critical for feature-level cost tracking

    -- Application Context
    tool_name VARCHAR(255) NOT NULL, -- e.g., 'hr_helpdesk', 'contract_review', 'supply_chain_forecast'
    feature_name VARCHAR(255), -- e.g., 'escalation_triage', 'clause_extraction'

    -- Compliance & PII Flags
    has_pii_detected BOOLEAN DEFAULT FALSE,
    pii_redacted_before_llm BOOLEAN DEFAULT FALSE,
    gdpr_pseudonymized_user_key VARCHAR(255), -- For Right to Erasure
    hipaa_compliant_route BOOLEAN DEFAULT FALSE,

    -- Response & Hashes
    response_hash BYTEA NOT NULL, -- Hash of the raw response
    response_preview TEXT, -- First 500 chars of response (for human auditing)
    full_prompt_text TEXT, -- THE TRICKY PART: See PII section below.
    full_response_text TEXT, -- Also tricky.

    -- System Metadata
    request_id VARCHAR(255),
    status_code INTEGER,
    latency_ms INTEGER
);

CREATE INDEX idx_audit_tenant_time ON llm_interaction_audit (tenant_id, event_timestamp);
CREATE INDEX idx_audit_user ON llm_interaction_audit (gdpr_pseudonymized_user_key);
CREATE INDEX idx_audit_tool ON llm_interaction_audit (tool_name);

from presidio_analyzer import AnalyzerEngine
from presidio_anonymizer import AnonymizerEngine
import json

analyzer = AnalyzerEngine()
anonymizer = AnonymizerEngine()

def redact_pii_from_text(text: str, user_id: str) -> tuple[str, dict]:
    """
    Analyzes text for PII, redacts it, and returns the safe text + mapping.
    The mapping is stored securely for possible re-injection post-LLM.
    """
    # Analyze for PII entities (PERSON, EMAIL_ADDRESS, PHONE_NUMBER, etc.)
    results = analyzer.analyze(text=text, language='en')

    # Create an anonymization config (replace with placeholders)
    anonymizer_config = {
        "DEFAULT": {"type": "replace", "new_value": "[REDACTED]"}
    }

    # Anonymize the text
    anonymized_result = anonymizer.anonymize(
        text=text,
        analyzer_results=results,
        operators=anonymizer_config
    )

    # The safe text to send to the LLM
    safe_text = anonymized_result.text

    # For context preservation, you might store a mapping.
    # e.g., "[REDACTED_1]": "John Doe". STORE THIS SECURELY, LINKED TO `user_id`.
    pii_map = {}
    for i, ann in enumerate(anonymized_result.items):
        if ann.operator == "replace":
            pii_map[f"[REDACTED_{i}]"] = ann.text

    return safe_text, pii_map


from openai import OpenAI
client = OpenAI()

@app.post("/ask-hr")
async def ask_hr_question(question: str, user: User):
    # 1. REDACT FIRST
    safe_prompt, pii_map = redact_pii_from_text(question, user.id)

    # 2. Log the event with PII flag
    audit_log = {
        "tenant_id": user.tenant,
        "user_id": user.id,
        "has_pii_detected": len(pii_map) > 0,
        "pii_redacted_before_llm": True,
        "prompt_hash": generate_hash(safe_prompt), # Hash the REDACTED prompt
        "full_prompt_text": safe_prompt, # Store the REDACTED version
        "tool_name": "hr_helpdesk"
    }
    # ... insert into llm_interaction_audit

    # 3. Send SAFE text to LLM
    try:
        response = client.chat.completions.create(
            model="gpt-4o",
            messages=[{"role": "user", "content": safe_prompt}]
        )
        llm_response = response.choices[0].message.content
    except openai.RateLimitError:
        # Real Error: openai.RateLimitError: You exceeded your current quota
        # Exact Fix: implement per-tenant rate limiting with Redis token bucket
        raise HTTPException(status_code=429, detail="Rate limit exceeded for tenant.")

    # 4. (Optional) Re-inject PII into the LLM response for user clarity
    final_response = llm_response
    for placeholder, original_value in pii_map.items():
        final_response = final_response.replace(placeholder, original_value)

    # 5. Log the response
    # ... log response_hash, cost, etc.
    return {"answer": final_response}

import hashlib
import json

def generate_immutable_log_entry(log_data: dict, previous_hash: str = None):
    """
    Creates a tamper-evident log entry.
    """
    # Create a deterministic string representation of the core data
    core_string = f"{log_data['tenant_id']}:{log_data['user_id']}:{log_data['prompt_text']}:{log_data['response_text']}"

    # Include the previous hash in the current hash to form a chain
    if previous_hash:
        string_to_hash = core_string + previous_hash
    else:
        string_to_hash = core_string

    # Generate the hash for this entry
    current_hash = hashlib.sha256(string_to_hash.encode('utf-8')).hexdigest()
    log_data['entry_hash'] = current_hash

    # Store the hash as BYTEA in PostgreSQL
    log_data['entry_hash_bytes'] = bytes.fromhex(current_hash)

    return log_data, current_hash

# Celery task to archive old logs to S3
import boto3
import pandas as pd
from sqlalchemy import create_engine
from datetime import datetime, timedelta

@app.task
def archive_old_audit_logs():
    engine = create_engine(DATABASE_URL)
    cutoff_date = datetime.utcnow() - timedelta(days=60)

    # Query logs older than 60 days
    query = f"SELECT * FROM llm_interaction_audit WHERE event_timestamp < '{cutoff_date.isoformat()}'"
    df = pd.read_sql_query(query, engine)

    if not df.empty:
        # Parquet is efficient for analytics
        file_key = f"audit-logs/archived_{cutoff_date.strftime('%Y%m')}.parquet"
        df.to_parquet(f"/tmp/{file_key}")

        s3 = boto3.client('s3')
        s3.upload_file(f"/tmp/{file_key}", 'your-compliance-bucket', file_key)

        # Optional: Delete archived rows from primary DB after successful upload
        # delete_query = f"DELETE FROM llm_interaction_audit WHERE event_timestamp < '{cutoff_date.isoformat()}'"
        # engine.execute(delete_query)

        print(f"Archived {len(df)} logs to S3.")

# audit_report.py
import pandas as pd
from datetime import datetime, timedelta

def generate_soc2_report(tenant_id: str, start_date: datetime, end_date: datetime):
    """Executes key auditor queries and returns a DataFrame."""
    query = f"""
    SELECT
        DATE(event_timestamp) as day,
        tool_name,
        user_id,
        model,
        SUM(prompt_tokens + response_tokens) as total_tokens,
        SUM(total_cost) as daily_cost,
        COUNT(*) as interaction_count
    FROM llm_interaction_audit
    WHERE tenant_id = '{tenant_id}'
        AND event_timestamp BETWEEN '{start_date}' AND '{end_date}'
        AND has_pii_detected = FALSE -- Demonstrate PII controls are working
    GROUP BY DATE(event_timestamp), tool_name, user_id, model
    ORDER BY day DESC;
    """
    df = pd.read_sql_query(query, engine)
    return df

def get_all_pii_interactions(tenant_id: str):
    """Auditor wants to see that PII detection is logged."""
    query = f"""
    SELECT event_timestamp, user_id, tool_name, prompt_preview
    FROM llm_interaction_audit
    WHERE tenant_id = '{tenant_id}'
        AND has_pii_detected = TRUE
    ORDER BY event_timestamp DESC
    LIMIT 100;
    """
    return pd.read_sql_query(query, engine)

# Run it
report_df = generate_soc2_report("acme_corp", datetime.utcnow() - timedelta(days=90), datetime.utcnow())
pii_df = get_all_pii_interactions("acme_corp")

print("## SOC2 Compliance Summary for Q1 ##")
print(report_df.to_string())
print("\n## PII Detection Audit Sample ##")
print(pii_df.to_string())

def validate_and_execute_sql(generated_sql: str, db_engine):
    # 1. Restrict to SELECT only
    if not generated_sql.strip().upper().startswith("SELECT"):
        raise ValueError("Only SELECT queries are permitted.")

    # 2. Validate with EXPLAIN to catch syntax errors or dangerous ops
    try:
        explain_query = f"EXPLAIN {generated_sql}"
        db_engine.execute(explain_query)
    except Exception as e:
        # Log this failure in your audit table!
        raise ValueError(f"Generated SQL failed validation: {e}")

    # 3. If valid, execute
    return pd.read_sql_query(generated_sql, db_engine)