Audit Logging¶

class ApprovalAuditLog:
    """Audit log for approval decisions."""

    def __init__(
        self,
        log_file: str | Path | None = None,
        max_memory_entries: int = 1000,
    ):
        ...

@dataclass
class AuditEntry:
    entry_id: str                    # Unique entry identifier
    timestamp: str                   # ISO 8601 timestamp of the decision
    request_id: str                  # Matching approval request ID
    action_type: str                 # Action type (e.g., "code", "final")
    risk_level: str                  # Risk level string (e.g., "high", "critical")
    approved: bool                   # Whether the action was approved
    status: str                      # ApprovalStatus value (e.g., "approved", "auto_denied")
    reason: str                      # Explanation for the decision
    approver: str                    # Who/what made the decision
    code_preview: str                # First 200 chars of the code (truncated)
    affected_resources: list[str]    # Resources impacted by the action
    metadata: dict[str, Any]         # Additional data (reversible, risk_reasons)

entry = AuditEntry.from_request_response(request, response)

entry_dict = entry.to_dict()
# Returns a plain dictionary suitable for JSON serialization

from rlm_code.rlm.approval import ApprovalAuditLog

# Memory-only audit log (no file persistence)
audit_log = ApprovalAuditLog(max_memory_entries=500)

# Persistent audit log (JSONL format)
audit_log = ApprovalAuditLog(
    log_file="logs/approval_audit.jsonl",
    max_memory_entries=1000,
)

from rlm_code.rlm.approval import ApprovalGate, ApprovalPolicy, ApprovalAuditLog

audit_log = ApprovalAuditLog(log_file="audit.jsonl")

gate = ApprovalGate(
    policy=ApprovalPolicy.CONFIRM_HIGH_RISK,
    audit_log=audit_log,
)

# All decisions through this gate are automatically logged
request = gate.check_action({"action": "code", "code": "os.remove('file.txt')"})
response = await gate.request_approval(request)
# Audit entry created automatically

from rlm_code.rlm.approval import (
    ApprovalAuditLog,
    ApprovalRequest,
    ApprovalResponse,
    ApprovalStatus,
    RiskAssessment,
    ToolRiskLevel,
)

audit_log = ApprovalAuditLog(log_file="audit.jsonl")

# Log is called automatically by ApprovalGate, but can be manual:
entry = audit_log.log(request, response)
print(f"Logged: {entry.entry_id} - {entry.status}")

# Get all entries
all_entries = audit_log.get_entries()

# Get last 10 entries
recent = audit_log.get_entries(limit=10)

# Get only approved entries
approved = audit_log.get_entries(approved_only=True)

# Get only denied entries
denied = audit_log.get_entries(denied_only=True)

# Filter by risk level
high_risk = audit_log.get_entries(risk_level="high")
critical = audit_log.get_entries(risk_level="critical")

denied_entries = audit_log.get_entries(denied_only=True, risk_level="critical")

for entry in denied_entries:
    print(f"[{entry.timestamp}] {entry.action_type}")
    print(f"  Risk: {entry.risk_level}")
    print(f"  Reason: {entry.reason}")
    print(f"  Code: {entry.code_preview}")
    print(f"  Resources: {', '.join(entry.affected_resources)}")
    print()

summary = audit_log.get_summary()

{
    "total": 150,
    "approved": 120,
    "denied": 30,
    "approval_rate": 0.8,
    "by_risk_level": {
        "safe": {"total": 80, "approved": 80, "denied": 0},
        "low": {"total": 30, "approved": 28, "denied": 2},
        "medium": {"total": 25, "approved": 10, "denied": 15},
        "high": {"total": 12, "approved": 2, "denied": 10},
        "critical": {"total": 3, "approved": 0, "denied": 3},
    },
}

{
    "total": 0,
    "approved": 0,
    "denied": 0,
    "approval_rate": 0.0,
    "by_risk_level": {},
}

audit_log.export_report("reports/compliance_report.md")

audit_log = ApprovalAuditLog(log_file="audit.jsonl")

# Load previous entries from file
loaded_count = audit_log.load_from_file()
print(f"Loaded {loaded_count} entries from file")

# Now get_entries() includes loaded entries
all_entries = audit_log.get_entries()

# Clear in-memory entries (file is NOT affected)
audit_log.clear()

# After clear, get_entries() returns empty
entries = audit_log.get_entries()  # []

# But the file still contains all previous entries
# You can reload them:
loaded = audit_log.load_from_file()

from rlm_code.rlm.approval import (
    ApprovalGate,
    ApprovalPolicy,
    ApprovalAuditLog,
    ConsoleApprovalHandler,
    RiskAssessor,
)
from rlm_code.rlm.approval.policy import RiskRule, ToolRiskLevel


# 1. Set up audit log with file persistence
audit_log = ApprovalAuditLog(
    log_file="logs/session_audit.jsonl",
    max_memory_entries=2000,
)

# 2. Set up risk assessor with custom rules
assessor = RiskAssessor()
assessor.add_rule(RiskRule(
    name="env_var_modification",
    pattern=r"os\.environ\[",
    risk_level=ToolRiskLevel.MEDIUM,
    reason="Environment variable modification detected",
    reversible=True,
))

# 3. Set up gate with all components
gate = ApprovalGate(
    policy=ApprovalPolicy.CONFIRM_MEDIUM_AND_UP,
    risk_assessor=assessor,
    approval_handler=ConsoleApprovalHandler(timeout_seconds=120).handle,
    audit_log=audit_log,
)


# 4. Agent execution loop
async def agent_loop(actions):
    for action in actions:
        request = gate.check_action(action)
        if request.requires_approval:
            response = await gate.request_approval(request)
            if not response.approved:
                print(f"Skipping: {response.reason}")
                continue
        # Execute approved action
        execute(action)


# 5. After execution, analyze the audit trail
summary = audit_log.get_summary()
print(f"Total decisions: {summary['total']}")
print(f"Approval rate: {summary['approval_rate']:.0%}")
print(f"Denied high-risk: {summary['by_risk_level'].get('high', {}).get('denied', 0)}")

# 6. Query specific entries
denied_critical = audit_log.get_entries(denied_only=True, risk_level="critical")
for entry in denied_critical:
    print(f"DENIED CRITICAL: {entry.code_preview}")
    print(f"  Reason: {entry.reason}")
    print(f"  Resources: {entry.affected_resources}")

# 7. Export compliance report
audit_log.export_report("reports/session_compliance.md")