ai-audit-trail

from ai_audit import (    AuditConfig, init_audit_config,    ReceiptCollector, ReceiptStore, ReceiptAction,    verify_chain, get_verify_key_hex,)
# 1. Configure once at startupinit_audit_config(AuditConfig(is_production=False))store = ReceiptStore()
# 2. Wrap every AI requestcollector = ReceiptCollector(trace_id="req-1", tenant_id="acme")collector.set_input("What is our GDPR policy?")collector.add_check("safety", score=0.02, threshold=0.8, fired=False)collector.set_output("Our GDPR policy states that...")collector.set_action(ReceiptAction.ALLOW)collector.emit(store)collector.cleanup()
# 3. Verify tamper-evidenceresult = verify_chain(store.get_by_tenant("acme"), get_verify_key_hex())assert result.valid  # Ed25519 + SHA-256 + hash-chain verified

from ai_audit import PiiConfig, PiiMode, PiiType
config = PiiConfig(    enabled_types=frozenset({PiiType.EMAIL, PiiType.PHONE, PiiType.IP}),    mode=PiiMode.REDACT,  # or HASH (SHA-256) or MASK (a***m))collector = ReceiptCollector(tenant_id="acme", pii_config=config)

from ai_audit.shredding import AESGCMDEKStore, encrypt_field, shred_tenant
dek_store = AESGCMDEKStore()dek_store.create_dek("tenant-acme")
field = encrypt_field("sensitive PII", dek_store, "tenant-acme")shred_tenant("tenant-acme", dek_store)  # Key destroyed — data unrecoverable# Hash-chain remains mathematically intact (hashes ciphertext, not plaintext)

from ai_audit.crosswalk import build_crosswalk, nist_function_map
crosswalk = build_crosswalk(receipts, chain_intact=True)for entry in crosswalk:    print(f"[{entry.status}] {entry.framework} {entry.control_id} — {entry.control_name}")
nist = nist_function_map(receipts)print(nist["GOVERN"].coverage)   # 0.0–1.0print(nist["MEASURE"].status)    # PASS / PARTIAL / FAIL

from ai_audit.report import ComplianceReportGenerator
gen = ComplianceReportGenerator(summary, verify_key_hex=get_verify_key_hex())gen.to_markdown()   # Documentation portalsgen.to_json()       # Automated pipelinesgen.to_html()       # Air-gapped servers

from ai_audit.export import export_evidence_package, verify_evidence_package
export_evidence_package(receipts, verify_key_hex, signing_key, "audit_2026.zip")# Bundle: receipts.jsonl + chain_metadata.json + public_key.hex + manifest.json (signed) + verify.py
# Auditor verifies offline:# python -m ai_audit verify audit_2026.zip

from ai_audit.sprt import SPRTMonitor
monitor = SPRTMonitor(tenant_id="acme")for receipt in receipts:    state = monitor.update(is_reject=(receipt.action == "reject"))print(state.status)      # CERTIFIED | MONITORING | FLAGGEDprint(state.confidence)  # 0.0–1.0

from ai_audit.drift import DriftMonitor
monitor = DriftMonitor(window_size=100)for receipt in receipts:    state = monitor.update(receipt.action.value)print(state.status)       # STABLE | DRIFTING | CRITICAL_DRIFTprint(state.drift_score)  # JSD 0.0–1.0

from ai_audit.contracts import BehavioralContract, Constraint, ContractMonitor
contract = BehavioralContract(    contract_id="safety-v1",    constraints=[        Constraint(name="no_pii_leak", kind="hard", field="action", operator="!=", value="reject"),        Constraint(name="quality", kind="soft", field="checks.safety.score", operator="<=", value=0.1, delta=0.5),    ],)monitor = ContractMonitor(contract)for receipt in receipts:    state = monitor.evaluate(receipt)
print(state.p)                   # Hard constraint satisfaction probabilityprint(state.delta)               # Maximum soft deviation observedprint(state.k)                   # Recovery steps after last violationprint(state.reliability_index)   # Theta: single compliance metric (0.0–1.0)print(state.status)              # COMPLIANT | DEGRADED | VIOLATED

from ai_audit.toolcall import seal_tool_call, verify_tool_call_chain
receipt = seal_tool_call(    agent_id="researcher",    tool_name="web_search",    tool_args={"query": "EU AI Act compliance"},    tool_result="Found 5 relevant documents...",    private_key=signing_key,    tenant_id="acme",)assert receipt.verify(signing_key.verify_key)

from ai_audit.tracegraph import TraceGraph
graph = TraceGraph(trace_id="workflow-1", tenant_id="acme")root = graph.add_node(agent_id="orchestrator", action="plan")graph.add_node(agent_id="researcher", action="search", parent_id=root.node_id)graph.add_node(agent_id="writer", action="draft", parent_id=root.node_id)
assert graph.verify_integrity()  # Hash-based tamper detectionassert not graph.has_cycles()    # DAG validationlineage = graph.get_agent_lineage(leaf_node.node_id)  # Root-to-leaf trace

from ai_audit.provenance import ProvenanceChain, ProvenanceRecord, SourceType
chain = ProvenanceChain(receipt_id="r1", tenant_id="acme")chain.add(ProvenanceRecord(source_type=SourceType.SYSTEM, source_id="prompt", trust_level=1.0, content_hash="..."))chain.add(ProvenanceRecord(source_type=SourceType.DOCUMENT, source_id="doc-123", trust_level=0.8, content_hash="..."))chain.add(ProvenanceRecord(source_type=SourceType.UNKNOWN, source_id="???", trust_level=0.0, content_hash="..."))
summary = chain.trust_summary()print(summary.system_grounded)       # True — has SYSTEM sourceprint(summary.potentially_injected)  # True — has UNKNOWN sourceprint(summary.min_trust)             # 0.0 — weakest link

from ai_audit.batch import MerkleBatcher
batcher = MerkleBatcher(tenant_id="acme", private_key=key, max_batch_size=2048)for receipt in receipts:    seal = batcher.add(receipt.receipt_id, receipt.seal_payload())    if seal:  # Auto-flushed at 2048 receipts        print(f"Batch sealed: {seal.merkle_root[:16]}...")
assert batcher.verify_chain_of_roots(key.verify_key)

from ai_audit.epochs import EpochManager
mgr = EpochManager(tenant_id="acme", private_key=key, max_epoch_size=10_000)for receipt in receipts:    seal = mgr.add_receipt(receipt)  # Auto-seals at 10kmgr.seal_epoch()                     # Or explicit rolloverassert mgr.verify_epoch_chain(key.verify_key)

from ai_audit.buffer import AuditBuffer, AuditBufferFullError
buffer = AuditBuffer(maxsize=50_000)  # ~5 seconds at 10k req/stry:    buffer.put(receipt)except AuditBufferFullError:    # Backpressure — reject the request rather than lose audit data    passbatch = buffer.drain(max_items=2048)

from ai_audit.storage import StorageBackend, InMemoryBackend
# Use the reference implementation for dev/testbackend = InMemoryBackend(max_receipts=50_000)
# Or implement your own:class PostgresBackend(StorageBackend):    def write_receipt(self, receipt): ...    def read_receipt(self, receipt_id): ...    def query_by_tenant(self, tenant_id, limit=100): ...    def healthcheck(self) -> bool: ...

# pip install "ai-audit-trail[otel]"from ai_audit.telemetry import record_seal, record_append, record_drift
record_seal(duration_seconds=0.000045, tenant_id="acme")record_append(tenant_id="acme", async_mode=True)record_drift(score=0.03, tenant_id="acme")

init_audit_config(AuditConfig(is_production=True, signing_key_hex="your-64-char-hex-key"))

from ai_audit import KeyProvider, init_key_provider
class VaultKeyProvider(KeyProvider):    def get_signing_key(self) -> nacl.signing.SigningKey:        secret = vault_client.secrets.kv.read_secret("secret/ai-audit/key")        return nacl.signing.SigningKey(bytes.fromhex(secret["data"]["key"]))    def get_verify_key_hex(self) -> str:        return self.get_signing_key().verify_key.encode().hex()
init_key_provider(VaultKeyProvider())

import redisstore = ReceiptStore(redis_client=redis.Redis(), use_lua=True)  # Lua mode: 10k+ req/s