Compliance Logging for AI Agents: Audit Trails for Regulated Industries

Why Standard Logging Is Not Enough for Regulated Industries

When an AI agent operates in healthcare, finance, or legal domains, every interaction becomes a potential audit target. Regulators and auditors need to answer questions like: What exactly did the agent tell the patient? What data did the agent access to make that recommendation? Who approved the agent's access to financial records? When was the agent's prompt last modified?

Standard application logs — designed for debugging and monitoring — fail compliance requirements in several ways. They can be modified or deleted. They lack cryptographic integrity verification. They do not enforce data retention schedules. And they often either log too much sensitive data or too little context for audit reconstruction.

Compliance logging is a separate, purpose-built layer that captures agent interactions with immutability guarantees, retention controls, and access audit capabilities.

The Compliance Audit Record

Design an audit record structure that captures everything an auditor needs to reconstruct what happened, why, and who was involved.

from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
import uuid
import hashlib
import json

class AuditAction(Enum):
    CONVERSATION_STARTED = "conversation_started"
    USER_MESSAGE_RECEIVED = "user_message_received"
    AGENT_RESPONSE_GENERATED = "agent_response_generated"
    TOOL_INVOKED = "tool_invoked"
    DATA_ACCESSED = "data_accessed"
    DATA_MODIFIED = "data_modified"
    PROMPT_CHANGED = "prompt_changed"
    MODEL_CHANGED = "model_changed"
    ACCESS_GRANTED = "access_granted"
    ACCESS_REVOKED = "access_revoked"
    CONVERSATION_ENDED = "conversation_ended"

@dataclass
class AuditRecord:
    id: str = field(default_factory=lambda: str(uuid.uuid4()))
    timestamp: str = field(default_factory=lambda: datetime.utcnow().isoformat() + "Z")
    action: AuditAction = AuditAction.CONVERSATION_STARTED
    actor_id: str = ""           # User ID or system ID that triggered the action
    actor_type: str = ""         # "user", "agent", "admin", "system"
    conversation_id: str = ""
    agent_name: str = ""
    resource_type: str = ""      # "patient_record", "financial_data", etc.
    resource_id: str = ""
    details: dict = field(default_factory=dict)
    data_classification: str = "" # "public", "internal", "confidential", "restricted"
    integrity_hash: str = ""

    def compute_hash(self, previous_hash: str = "") -> str:
        """Chain hash for tamper detection."""
        payload = json.dumps({
            "id": self.id,
            "timestamp": self.timestamp,
            "action": self.action.value,
            "actor_id": self.actor_id,
            "conversation_id": self.conversation_id,
            "details": self.details,
            "previous_hash": previous_hash,
        }, sort_keys=True)
        self.integrity_hash = hashlib.sha256(payload.encode()).hexdigest()
        return self.integrity_hash

Immutable Log Storage

Compliance logs must be tamper-evident. Use a combination of hash chaining and write-once storage to ensure integrity.

from sqlalchemy import Column, String, DateTime, Text, Index
from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy.orm import declarative_base

Base = declarative_base()

class AuditLog(Base):
    __tablename__ = "audit_log"

    id = Column(String, primary_key=True)
    timestamp = Column(DateTime, nullable=False, index=True)
    action = Column(String, nullable=False, index=True)
    actor_id = Column(String, nullable=False, index=True)
    actor_type = Column(String, nullable=False)
    conversation_id = Column(String, nullable=False, index=True)
    agent_name = Column(String, nullable=False)
    resource_type = Column(String, default="")
    resource_id = Column(String, default="")
    details = Column(JSONB, default={})
    data_classification = Column(String, nullable=False)
    integrity_hash = Column(String, nullable=False)
    previous_hash = Column(String, nullable=False, default="")

    __table_args__ = (
        Index("idx_audit_conversation", "conversation_id", "timestamp"),
        Index("idx_audit_actor", "actor_id", "timestamp"),
        Index("idx_audit_resource", "resource_type", "resource_id", "timestamp"),
    )


class ComplianceLogger:
    def __init__(self, db_session):
        self.db = db_session
        self.last_hash = ""

    async def log(self, record: AuditRecord):
        """Write an immutable, hash-chained audit record."""
        record.compute_hash(self.last_hash)

        log_entry = AuditLog(
            id=record.id,
            timestamp=datetime.fromisoformat(record.timestamp.rstrip("Z")),
            action=record.action.value,
            actor_id=record.actor_id,
            actor_type=record.actor_type,
            conversation_id=record.conversation_id,
            agent_name=record.agent_name,
            resource_type=record.resource_type,
            resource_id=record.resource_id,
            details=record.details,
            data_classification=record.data_classification,
            integrity_hash=record.integrity_hash,
            previous_hash=self.last_hash,
        )

        self.db.add(log_entry)
        await self.db.commit()
        self.last_hash = record.integrity_hash

Instrumenting the Agent for Compliance

Wrap agent operations to emit audit records at every decision point.

compliance_logger = ComplianceLogger(db_session)

async def compliant_agent_run(user_message: str, user_id: str, conversation_id: str):
    # Log conversation start
    await compliance_logger.log(AuditRecord(
        action=AuditAction.CONVERSATION_STARTED,
        actor_id=user_id,
        actor_type="user",
        conversation_id=conversation_id,
        agent_name="healthcare-agent",
        data_classification="confidential",
    ))

    # Log user message (redact PHI in details)
    await compliance_logger.log(AuditRecord(
        action=AuditAction.USER_MESSAGE_RECEIVED,
        actor_id=user_id,
        actor_type="user",
        conversation_id=conversation_id,
        agent_name="healthcare-agent",
        details={"message_length": len(user_message), "contains_phi": detect_phi(user_message)},
        data_classification="restricted" if detect_phi(user_message) else "confidential",
    ))

    response = await agent.run(user_message)

    # Log data access if tools were used
    for tool_call in (response.tool_calls or []):
        await compliance_logger.log(AuditRecord(
            action=AuditAction.DATA_ACCESSED,
            actor_id="healthcare-agent",
            actor_type="agent",
            conversation_id=conversation_id,
            agent_name="healthcare-agent",
            resource_type=classify_resource(tool_call.function.name),
            resource_id=extract_resource_id(tool_call.function.arguments),
            details={"tool_name": tool_call.function.name},
            data_classification="restricted",
        ))

    # Log agent response
    await compliance_logger.log(AuditRecord(
        action=AuditAction.AGENT_RESPONSE_GENERATED,
        actor_id="healthcare-agent",
        actor_type="agent",
        conversation_id=conversation_id,
        agent_name="healthcare-agent",
        details={
            "response_length": len(response.content),
            "model": response.model,
            "tools_used": [tc.function.name for tc in (response.tool_calls or [])],
        },
        data_classification="confidential",
    ))

    return response.content

Data Retention and Archival

Different regulations have different retention requirements. HIPAA requires 6 years. SOC2 typically requires 1 year. Build a retention policy engine that enforces these rules.

from datetime import timedelta

RETENTION_POLICIES = {
    "hipaa": timedelta(days=2190),      # 6 years
    "soc2": timedelta(days=365),        # 1 year
    "financial": timedelta(days=2555),  # 7 years
    "default": timedelta(days=365),
}

async def archive_expired_logs(db_session, policy: str = "hipaa"):
    """Move expired logs to cold storage and delete from primary."""
    retention = RETENTION_POLICIES.get(policy, RETENTION_POLICIES["default"])
    cutoff = datetime.utcnow() - retention

    # Export to S3 / GCS cold storage first
    expired_logs = await db_session.execute(
        text("""
            SELECT * FROM audit_log
            WHERE timestamp < :cutoff
            ORDER BY timestamp
        """),
        {"cutoff": cutoff},
    )

    rows = expired_logs.fetchall()
    if rows:
        archive_path = await export_to_cold_storage(rows, policy)

        # Only delete after confirming archive write
        await db_session.execute(
            text("DELETE FROM audit_log WHERE timestamp < :cutoff"),
            {"cutoff": cutoff},
        )
        await db_session.commit()

        return {"archived": len(rows), "path": archive_path}
    return {"archived": 0}

Integrity Verification

Periodically verify that the hash chain has not been broken, which would indicate tampering.

async def verify_audit_integrity(db_session, conversation_id: str = None) -> dict:
    """Verify hash chain integrity for audit logs."""
    query = "SELECT * FROM audit_log"
    params = {}
    if conversation_id:
        query += " WHERE conversation_id = :cid"
        params["cid"] = conversation_id
    query += " ORDER BY timestamp ASC"

    result = await db_session.execute(text(query), params)
    logs = result.fetchall()

    if not logs:
        return {"status": "empty", "records_checked": 0}

    previous_hash = ""
    broken_at = None

    for log in logs:
        record = AuditRecord(
            id=log.id,
            timestamp=log.timestamp.isoformat() + "Z",
            action=AuditAction(log.action),
            actor_id=log.actor_id,
            conversation_id=log.conversation_id,
            details=log.details,
        )
        expected_hash = record.compute_hash(previous_hash)

        if expected_hash != log.integrity_hash:
            broken_at = log.id
            break
        previous_hash = log.integrity_hash

    return {
        "status": "valid" if not broken_at else "tampered",
        "records_checked": len(logs),
        "broken_at_record": broken_at,
    }

FAQ

What is the difference between compliance logging and regular application logging?

Regular logging serves developers — it captures debug information, error traces, and performance data. Compliance logging serves auditors and regulators — it captures who did what, when, to which data, and with what authorization. Compliance logs must be immutable (tamper-evident), retained for specific periods, access-controlled, and complete enough to reconstruct any interaction. They run as a separate system with stricter write guarantees than application logs.

Do I need to store the full LLM prompt and response in audit logs?

It depends on your regulatory framework. HIPAA requires that you can reconstruct what information was accessed and disclosed, which means storing enough detail to reproduce the interaction. However, storing full prompts containing PHI creates its own compliance burden — the audit log itself becomes protected health information. A common approach is to store the full interaction in an encrypted, access-controlled archive and keep only metadata (token counts, tool names, classification labels) in the primary audit log.

How do I handle audit logging when the agent accesses data across multiple compliance domains?

Tag each audit record with the highest applicable data classification. If a single tool call accesses both patient records (HIPAA) and payment information (PCI-DSS), classify the record as both and apply the stricter retention policy. Maintain a resource classification registry that maps tool names and data sources to compliance domains, and update it whenever you add new tools or data sources to the agent.

---

#Compliance #AuditLogging #HIPAA #SOC2 #AIAgents #RegulatedIndustries #AgenticAI #LearnAI #AIEngineering