Audit pipeline

AuditPipeline lives in engine/core/src/engine.rs (lines 68–165). It is non-blocking — every audit write is spawned as a Tokio task so it cannot affect request tail latency.

record

impl AuditPipeline {
    pub fn record(&self, event: &str, meta: &serde_json::Value);
}

The first argument is the event-type tag (string). The pipeline maps it to an AuditEventType and forwards both to the bound sink:

pub enum AuditEventType {
    ProtocolInvocation,
    CapabilityGrant,
    AuthorizationCheck,
    RateLimitExceeded,
    CircuitBreakerOpen,
    SecurityViolation,
    Error,
}

If the string does not match a known type, it is logged as AuditEventType::Error and metric audit_unknown_event increments — never silently dropped.

When the engine records

The 8-stage pipeline writes to the audit pipeline at known points:

Protocol modules can also record directly via ctx.audit().record(...). Most do not — the engine records on their behalf at stage 8.

Sinks

The pipeline owns a Box<dyn AuditLogger> and forwards every event to it. Production deployments bind ProductionAuditLogger (writes to structured tracing + Prometheus). Dev builds bind MemoryAuditLogger (in-process ring buffer for tests).

#[async_trait]
pub trait AuditLogger: Send + Sync {
    async fn log(&self, event: AuditEventType, meta: serde_json::Value) -> Result<(), AuditError>;
}

Implementations include:

• ProductionAuditLogger — emits to tracing + metrics + the hash-chained store (MAX)
• KafkaAuditLogger — fans events out to Kafka for downstream consumers (feature-gated kafka-sink)
• MemoryAuditLogger — in-memory ring; used by tests
• CompositeAuditLogger — fan-out wrapper for multi-sink deployments

Hash chaining via MAX

The production sink chains every audit record into MAX::audit_log_entry so the resulting chain is verifiable end-to-end. The audit-chain head hash is returned in response metadata so clients can store it and later replay via MAX::traceability_graph.

Write semantics

fn record(&self, event: &str, meta: &Value) {
    let event_type = parse_event_type(event);
    let meta = meta.clone();
    let logger = self.logger.clone();
    tokio::spawn(async move {
        if let Err(e) = logger.log(event_type, meta).await {
            tracing::warn!(error = %e, "audit log write failed");
        }
    });
}

The implementation is fire-and-forget for the fast path. If the logger errors, the warning is traced but the request is not affected. The chain integrity is reconciled in the background by MAX::compliance_report.

Backpressure

When the underlying sink falls behind (e.g., Kafka unreachable), the production audit pipeline switches to a disk-backed buffer at $MAP_AUDIT_BUFFER_DIR (default /var/lib/map/audit/). The buffer is drained when the sink recovers. Buffer overflow triggers RateLimited at Stage 3 for all but map.*.read operations — the engine refuses to accept new state-changing work it cannot audit.

Tests

engine/core/src/tests.rs::audit_pipeline_async verifies:

• writes never block the response path (record returns < 100µs)
• failed writes log but do not panic
• event-type parsing falls back to Error for unknown strings
• the disk-backed buffer drains correctly after a simulated sink outage

See also

• MAX protocol — the hash-chained record itself
• Observability — traces, metrics, and the OTEL surface
• Governance — Refusal carries reasons