vaxelia-ai-core

Shared compliance-reporting core for the Vaxelia AI-provider SDK wrappers (vaxelia-ai-anthropic, and future provider wrappers). It provides a single ComplianceReporter that ships every AI decision to your tenant's decision-log endpoint and, when delivery fails, durably buffers the decision to an encrypted local disk store for later retry — so an AI decision is never silently lost.

This package is an external-integrator library: it has no internal-platform dependencies and never instantiates a logger or telemetry client. You inject a logger and (optionally) an HTTP transport. The only runtime dependency is cryptography (for AES-256-GCM).

The JSON body posted to the tenant API is byte-compatible with the TypeScript @vaxelia/ai-core SDK (the same camelCase keys), so both SDK families target the same decision-log endpoint.

Install

pip install vaxelia-ai-core

Requires Python >= 3.9.

Configuration

Every ReporterConfig field is optional and falls back to an environment variable, in keeping with 12-factor configuration. Explicit config always wins over the environment.

Additional config (no env fallback): buffer_dir, buffering_enabled, logger, transport, sleep, max_attempts, base_delay_seconds.

Fail-closed buffer key

When buffering is enabled (the default), a valid buffer key must be resolvable — either via config.buffer_key or VAXELIA_BUFFER_KEY. If none is present, the constructor raises MissingBufferKeyError; if the key does not decode to exactly 32 bytes, it raises InvalidBufferKeyError. This is deliberate: rather than risk writing plaintext PII to disk, the reporter refuses to start. If you do not want a disk buffer, set buffering_enabled=False — then no key is required, but undeliverable decisions hard-fail instead of being buffered.

Generate a key:

python -c "import os, base64; print(base64.b64encode(os.urandom(32)).decode())"

Resilience behavior

• Encryption at rest — each buffered decision is stored as a separate file containing nonce | ciphertext+tag and encrypted with AES-256-GCM using a fresh random 12-byte nonce per entry. Files are written with mode 0o600.
• Buffer directory — defaults to <tempdir>/vaxelia-ai-buffer; override with buffer_dir.
• Exponential backoff — transient failures (network error, HTTP 5xx, HTTP 429) are retried with delay min(base_delay_seconds * 2^(attempt-1), 30s) (± jitter) up to max_attempts total attempts. Backoff only follows a transient failure and resets after a success, so a healthy endpoint drains a buffered backlog with no added inter-entry delay.
• 7-day local retention — on every flush, buffered entries older than seven days are purged and a WARN is logged through the injected logger (visible, not silent).
• Hard-fail, not silent drop — if a decision can neither be delivered nor durably buffered, log_decision raises DecisionNotRecordedError.
• Permanent rejection — an HTTP 4xx other than 429 is permanent: the decision is not retried or buffered, and log_decision raises PermanentRejectionError.
• No secrets in errors or logs — the API key and buffer key values never appear in raised errors or logged fields.

Usage

import json
from datetime import datetime, timezone

from vaxelia_ai_core import AIDecision, ComplianceReporter, ReporterConfig


def logger(level, msg, fields):
    print(json.dumps({"level": level, "msg": msg, **dict(fields)}))


reporter = ComplianceReporter(
    ReporterConfig(
        # omit any field to use the matching env var
        tenant_api_url="https://your-tenant.example/decision-log",
        api_key="...",       # or VAXELIA_API_KEY
        buffer_key="...",    # 32-byte base64, or VAXELIA_BUFFER_KEY
        logger=logger,
    )
)

decision = AIDecision(
    ai_system_id="ai-sys-prod-1",
    model_used="claude-some-model",
    input={"prompt": "Summarize the contract."},
    output={"text": "The contract obligates..."},
    status="completed",
    decided_at=datetime.now(timezone.utc).isoformat(),
)

# Delivers to the tenant API; buffers encrypted on transient failure.
reporter.log_decision(decision)

# Periodically (e.g. on a timer or at shutdown) drain the buffer.
result = reporter.flush_buffer()
print(f"flushed {result['sent']}, {result['remaining']} still buffered")

Injecting a transport (for tests)

The default transport uses the stdlib urllib.request over HTTPS. For tests, inject a callable (url, body: bytes, headers) -> HttpResponse that records calls and returns scripted responses (or raises to simulate a network error). You can also inject sleep to make backoff deterministic.

Development

python -m venv .venv
.venv/bin/pip install -e ".[test]"
.venv/bin/python -m pytest
.venv/bin/python -m build   # sdist + wheel into dist/