Source Code Reading: AuditService — Cross-Process Audit Kafka Consumer

“

Series: S9 Source Code Reading · Article 17 | Level: Advanced | Reading Time: 25 min

#TL;DR

AuditService is the audit engine within coomia-dip's Metadata & Governance layer (Metadata & Governance Layer, merged into Control Layer), responsible for cross-process operation auditing, decision tracing, and compliance recording. It exposes 10 RPC methods via gRPC covering four capability domains: event recording (RecordEvent/BatchRecordEvents), log querying (QueryAuditLogs/GetEntityHistory), decision tracing (GetDecisionTrace/GetDecisionHistory), and data export (ExportAuditLogs). This article dissects the 13 audit event types in the Proto contract, AuditEmitter's dual-channel emission strategy (gRPC-first + JSONL fallback), ComputeAuditLogger's Kafka three-level degradation mechanism, and DecisionTrace's input snapshot and reasoning step recording model.

#Table of Contents

1. Overall Architecture: Convergence Point for Multi-Source Audit Events
2. Proto Contract: 13 Audit Event Types
3. AuditEvent Data Model: 16-Field Compliance Design
4. AuditEmitter: Dual-Channel Emission Strategy
5. ComputeAuditLogger: Kafka Three-Level Degradation
6. DecisionTrace: Decision Tracing and Explainability
7. QueryAuditLogs: Multi-Dimension Filtered Queries
8. Sensitive Data Masking: _sanitize Function
9. Data Lifecycle: CleanupOldEvents
10. Export and Compliance: ExportAuditLogs
11. Key Takeaways

#1. Overall Architecture: Convergence Point for Multi-Source Audit Events

AuditService is the convergence point for audit events from all Layers, receiving events from three directions:

Reasoning & Decision Layer (Reasoning)  --> ComputeAuditLogger  --> Kafka topic / gRPC
Agent Runtime Layer (Actions)    --> AuditEmitter         --> gRPC / JSONL fallback
Data Layer (Data Ops)   --> Direct gRPC call     --> AuditService

intelligence-Layer/src/
├── reasoning_decision_plane/security/
│   └── compute_audit.py              # Reasoning & Decision Layer audit logger
├── agent_runtime_plane/action/
│   └── audit_emitter.py              # Agent Runtime Layer audit emitter
proto/plane_g/
└── audit_service.proto               # 298-line audit contract

Design philosophy: Audit event recording must be "fire-and-forget" -- it must never block or fail the main business flow due to audit system failures.

#2. Proto Contract: 13 Audit Event Types

AuditService defines 13 event types covering the full lifecycle, organized into four logical groups: CRUD group (CREATE/UPDATE/DELETE/QUERY), Decision group (REASONING/DECISION/APPROVED/REJECTED/EXECUTED/ACTION), and Compliance group (ACCESS/EXPORT/MASKING).

The Decision group's 6 types completely record the decision lifecycle: REASONING -> DECISION -> APPROVED/REJECTED -> EXECUTED/ACTION.

#3. AuditEvent Data Model: 16-Field Compliance Design

The AuditEvent message contains 16 fields balancing business auditing and compliance requirements:

message AuditEvent {
  string log_id = 1;
  AuditEventType event_type = 2;
  string entity_id = 3;
  string user_id = 4;
  string world_id = 5;
  google.protobuf.Timestamp timestamp = 6;
  string trace_id = 7;
  AuditEventStatus status = 8;
  string summary = 9;
  google.protobuf.Struct details = 10;
  string decision_id = 11;
  string tenant_id = 12;
  string data_classification = 13;
  string access_purpose = 14;
  string client_ip = 15;
  string user_agent = 16;
}

Compliance fields (13-16) are designed for GDPR/SOX requirements: data_classification marks accessed data level, access_purpose records the access purpose (GDPR's "purpose limitation" principle), and client_ip + user_agent enable source tracing and anomaly detection.

#4. AuditEmitter: Dual-Channel Emission Strategy

AuditEmitter implements elegant dual-channel degradation for Agent Runtime Layer:

class AuditEmitter:
    def __init__(self, plane_g_endpoint: str | None = None,
                 local_log_path: str = "logs/audit_fallback.jsonl"):
        self._local_log_path = Path(local_log_path)
        if plane_g_endpoint:
            self._init_grpc(plane_g_endpoint)

Lazy Proto Stub loading uses double try/except since Proto stubs may not be compiled yet. The gRPC initialization attempts imports from two possible module paths (reasoning_decision_plane.generated and agent_runtime_plane.api.generated).

Dual-channel emission: The _emit method implements gRPC-to-JSONL degradation:

async def _emit(self, event: AuditEvent) -> None:
    """NEVER raises; all errors are logged as warnings."""
    try:
        if self._stub is not None:
            self._send_to_plane_g(event)
            return
    except Exception:
        logger.warning("Failed to send to Metadata & Governance Layer, falling back to local log")
    try:
        self._write_local_log(event)
    except Exception:
        logger.warning("Failed to write to local log")

Core design principle: NEVER raises. Even if both channels fail, only warnings are logged -- exceptions never propagate to callers.

#5. ComputeAuditLogger: Kafka Three-Level Degradation

ComputeAuditLogger implements three-level degradation for Reasoning & Decision Layer: Kafka -> gRPC -> Log Warning:

def flush(self) -> int:
    events = self.drain_events()
    if not events: return 0
    sent = self._send_kafka(events)
    if sent > 0: return sent
    sent = self._send_grpc(events)
    if sent > 0: return sent
    logger.warning("No external audit sink available, %d events dropped", len(events))
    return 0

Memory buffer design: Events are stored in-memory first (self._events), batch-retrieved and cleared via drain_events(). Kafka sending uses acks=1 and retries=1 -- "best effort" configuration where audit logs can be lost rather than blocking the computation path.

#6. DecisionTrace: Decision Tracing and Explainability

DecisionTrace is the most complex data model, providing complete decision explainability through three layers:

message DecisionTrace {
  string decision_id = 1;
  string commit_id = 3;                    // Nessie commit ID
  repeated InputSnapshot inputs = 6;        // Input snapshots
  repeated ReasoningStep reasoning_steps = 7;  // Reasoning steps
  repeated FiredRule fired_rules = 8;       // Fired rules
}

InputSnapshot freezes entity state at decision time. ReasoningStep records each reasoning step's input/output and confidence. FiredRule records each rule's contribution weight. These three layers enable complete post-hoc reconstruction: "what data was seen" -> "what reasoning occurred" -> "which rules participated" -> "what was the final decision."

#7. QueryAuditLogs: Multi-Dimension Filtered Queries

Audit log queries support 9 filter dimensions including entity_id, user_id, world_id, decision_id, time range, event types, and statuses. Cursor pagination uses page_token instead of offset, avoiding performance issues with large offsets.

#8. Sensitive Data Masking: _sanitize Function

AuditEmitter masks sensitive fields before recording audit events:

_SENSITIVE_FIELDS = frozenset({"password", "secret", "token", "api_key"})

def _sanitize(data: dict[str, Any] | None) -> dict[str, Any]:
    for key, value in data.items():
        if key.lower() in _SENSITIVE_FIELDS:
            result[key] = "***"
        elif isinstance(value, dict):
            result[key] = _sanitize(value)
        else:
            result[key] = value
    return result

Recursive masking: Nested dictionaries are processed recursively. frozenset ensures O(1) field name lookup. The _make_json_safe function handles types unsupported by Proto Struct (e.g., datetime converted to ISO format strings).

#9. Data Lifecycle: CleanupOldEvents

Audit log storage is finite. CleanupOldEvents provides retention-based cleanup with dry_run mode for previewing impact before actual deletion.

#10. Export and Compliance: ExportAuditLogs

Audit logs support three export formats: JSON, CSV, and Parquet. Parquet support is designed for large-scale analytics -- compliance teams can export audit logs as Parquet files for interactive analysis in Spark/Trino.

#11. Key Takeaways

1. Fire-and-forget principle: Audit recording must never block or fail the main business flow; all exceptions are caught and degraded.
2. Three-level degradation: Kafka -> gRPC -> JSONL ensures audit events are never completely lost under any infrastructure failure.
3. 13 event types comprehensively cover the full lifecycle from CRUD to decisions to compliance.
4. DecisionTrace three-layer structure: InputSnapshot -> ReasoningStep -> FiredRule provides complete decision explainability.
5. Recursive sensitive field masking: _sanitize ensures audit logs themselves don't leak sensitive information.
6. Cursor pagination: page_token replaces offset to avoid large-offset query performance issues.

#Next Article

S9-18: LineageService -- Entity + Field Level Lineage, where we dive into Metadata & Governance Layer's lineage tracking service to understand dual-perspective lineage graphs and impact analysis.

Tags: #coomia-dip #source-code-reading #audit-service #kafka-consumer #compliance #grpc #Layer-g