Forem: Jason Shotwell

Building Bilateral Receipts for AI Agent Actions

Jason Shotwell — Mon, 20 Apr 2026 00:17:40 +0000

Your AI agent just approved a $75,000 loan. Can you prove who authorized it? Can you prove what the result was? Can you prove the policy that was active when the decision was made?

If your answer involves grepping log files, you have a problem. August 2026 is coming, and the EU AI Act requires tamper-evident records for high-risk AI systems. Not logs. Records.

I built a system that solves this. Here's how it works.

The Problem: Logging Is Not Proof

Most AI agent frameworks have some form of logging. LangChain has callbacks. CrewAI has verbose mode. OpenAI has the API response. But none of them answer the hard questions:

Was this action authorized before it ran? Logs record what happened. They don't prove what was supposed to happen.

Can you verify the record hasn't been tampered with? Anyone with write access to your log store can alter a record after the fact. During a regulatory audit, that's a problem.

Can a third party verify without your help? If the auditor needs your internal tooling to check a record, the verification isn't independent.

These aren't theoretical concerns. EU AI Act Article 12 requires record-keeping with integrity guarantees. Article 14 requires human oversight that's demonstrable. If your agent approved a high-stakes decision and a regulator asks for proof, "here are some log lines" won't cut it.

The Solution: Covenants + Bilateral Receipts

I added a system called Gate to AIR Blackbox that handles both sides: what agents are allowed to do (pre-execution) and what they actually did (post-execution), with cryptographic proof at every step.

It has three components:

1. Covenants — Policy as YAML

A covenant is a YAML file that declares the rules before the agent runs. Three rule types: permit, forbid, require_approval. Conditions are supported via when and unless.

agent: loan-processor
version: "1.0"
rules:

permit: read_credit_score
permit: calculate_risk
permit: approve_loan when: "amount <= 50000"
require_approval: approve_loan when: "amount > 50000"
forbid: delete_records
forbid: modify_credit_score Precedence is strict: forbid > require_approval > permit > default deny. If no rule matches, the action is denied. This is deliberate — fail closed, not open.

The covenant is SHA-256 hashed, and that hash is embedded in every receipt. Change one rule, and every subsequent receipt carries a different hash. An auditor can verify exactly which policy was active for any past action.

2. Bilateral Receipts — Two-Phase Proof

Every action produces a receipt with two cryptographic phases:

Phase 1 (Authorization): The gate evaluates the covenant, makes a decision, and signs the authorization with Ed25519. The action payload is SHA-256 hashed — raw data (PII, financial details) never enters the receipt.

Phase 2 (Seal): After execution, the result is hashed and sealed into the same receipt with a second Ed25519 signature. The seal covers the authorization signature, binding the entire lifecycle.

from air_blackbox.gate import Gate, Covenant

covenant = Covenant.from_yaml("covenant.yaml")
gate = Gate(covenant=covenant)

Phase 1: Authorization

receipt = gate.authorize(
agent_id="loan-processor",
action_name="approve_loan",
payload={"applicant": "jane@example.com", "amount": 75000},
context={"amount": 75000},
)

if receipt.authorized:
result = process_loan(...)

# Phase 2: Seal

gate.seal(receipt, result=result, status="success")

Any third party can verify with just the public key

report = gate.verify(receipt)
print(report["overall"]) # True
A single receipt answers all three hard questions:

Was it authorized? The authorization signature proves the covenant was checked and the decision was made before execution.
Is it tamper-proof? Ed25519 signatures are asymmetric. Altering any field invalidates the signature. No shared secret needed to verify.
Can a third party verify? Give them the public key and the receipt. That's it.

3. HMAC-SHA256 Audit Chains

Individual receipts are strong. But an attacker could delete a receipt entirely. To prevent that, every receipt is also chained into an HMAC-SHA256 audit trail — each entry includes the hash of the previous entry. Delete or alter one, and every entry after it breaks.

This is the same principle behind blockchain, but without the overhead. No consensus, no network, no tokens. Just a hash chain stored locally.

Delegation Chains

The real world isn't one agent doing one thing. It's orchestrators delegating to sub-agents, which delegate to other sub-agents. Gate handles this with parent_receipt_id:

Orchestrator gets authorized

parent = gate.authorize("orchestrator", "delegate_task",
payload={"task": "send confirmation"})

Sub-agent links back to parent

child = gate.authorize("notifier", "send_email",
payload={"to": "jane@co.com"},
parent_receipt=parent)

Walk the chain back to the root

chain = gate.walk_delegation_chain(child)

[orchestrator_receipt, notifier_receipt]

Every receipt in the chain is independently verifiable. If a sub-agent misbehaves, the chain shows exactly who authorized the delegation and when.

Human Approval

When a covenant rule says require_approval, Gate pauses execution and calls your callback. You decide the interface — Slack, email, CLI prompt, whatever:

def slack_approval(receipt):
# Send to Slack, wait for button click
return ask_slack_channel(receipt)

gate = Gate(covenant=covenant, on_approval_needed=slack_approval)
If no callback is registered, require_approval defaults to deny. Fail closed.

The approval decision is signed into the receipt. A regulator can verify that a human was in the loop for that specific action.

What It Doesn't Do

Some things Gate deliberately avoids:

It doesn't store raw payloads. Only SHA-256 hashes. Your PII never enters the receipt.
It doesn't require a network. Everything runs locally. No cloud dependency.
It doesn't make legal compliance claims. It provides technical building blocks for audit-readiness. A covenant + receipt + chain is evidence, not certification.

Numbers

I stress-tested the system with 58 tests covering covenants, receipts, the gate engine, delegation chains, persistence, adversarial tampering, and edge cases.

Performance on Apple Silicon: 9,300+ authorizations per second, 3,500+ full lifecycles (authorize + seal + verify) per second with Ed25519 signing. A 100-deep delegation chain verifies correctly. A 50-rule covenant evaluates correctly.

Adversarial tests: swapping receipt IDs after signing, changing covenant hashes, flipping the authorized flag, altering the decision field — all detected by signature verification.

Try It

pip install air-blackbox[gate]
from air_blackbox.gate import Gate, Covenant

covenant = Covenant.from_yaml_string("""
agent: my-agent
version: "1.0"
rules:

permit: read
require_approval: write
forbid: delete """)

gate = Gate(covenant=covenant)
receipt = gate.authorize("my-agent", "read")
print(receipt.authorized) # True
print(gate.verify(receipt)["overall"]) # True
The full source is at github.com/airblackbox/airblackbox under sdk/air_blackbox/gate/. Example covenants for a loan processor and a browser automation agent are included.

What's Next

The covenant DSL today handles simple field-operator-value conditions. Next up: boolean logic (and/or), regex matching on action names, and rate-limit rules (e.g., "permit send_email unless more than 10 in the last hour").

The receipt format is designed for interoperability. The long-term goal is a published spec that any framework can implement — so receipts from a LangChain agent and a CrewAI agent chain together seamlessly.

If this is useful to you, star the repo. If you find a bug, open an issue. If you have a use case that doesn't fit, I want to hear about it.

This is not a certified compliance test. It is a technical building block for teams preparing for EU AI Act enforcement (August 2, 2026).

Three Questions Every EU AI Act Auditor Will Ask About Your Python AI Agent

Jason Shotwell — Wed, 15 Apr 2026 19:32:39 +0000

The EU AI Act's high-risk enforcement deadline is August 2, 2026. That is 109 days from today.

If your Python code runs an AI agent that makes decisions affecting someone's money, healthcare, job, housing, or insurance, those decisions are about to become legal records. The system producing them has to prove three things. This post walks through those three questions, why most Python AI codebases cannot answer them, and what a small community of open-source developers is building to close the gap.

I am one of those developers, and I want to be up front: the tool I am about to describe stands on work done by others. I will credit them throughout.

About the scanner

AIR Blackbox is the flight recorder for autonomous AI agents. Record, replay, enforce, audit. It is Apache 2.0, runs locally, and finishes a full scan in under ten seconds.

Install

pip install air-blackbox

Run your first scan

air-blackbox comply --scan .

Expected output (abbreviated)

Article 9  — Risk Management              PASS (3/3 checks)
Article 10 — Data Governance              WARN (2/3 checks)
Article 11 — Technical Documentation      PASS (4/4 checks)
Article 12 — Record-Keeping               FAIL (2/5 checks)
Article 13 — Transparency                 WARN (4/6 checks)
Article 14 — Human Oversight              PASS (3/3 checks)
Article 15 — Accuracy & Robustness        FAIL (3/6 checks)

Every FAIL and WARN line includes a fix hint pointing at the specific code location and the specific regulatory clause.

Now let us go through the three questions.

Question 1: Is this the same agent that acted yesterday?

An AI agent running a continuous decision loop, a while True, a scheduled runner, or a tick-based pattern is almost certainly executing in a different process today than yesterday. It reloaded memory from some store. It fetched its tool set. It started producing outputs.

How does a regulator know the agent producing today's loan denial is the same agent that was approved in last month's conformity assessment? How do you prove a compromised environment did not load tampered memory and produce a subtly different agent wearing the same name?

This is the agent identity continuity gap. The NIST RFI on AI Agent Security (Docket NIST-2025-0035) names it explicitly. The FINOS AI Governance Framework response to that RFI treats it as a primary unresolved problem.

The community is already solving this

Three open standards exist today, built by different people for different use cases, all interoperable:

air-trust: Ed25519 agent identity keys and HMAC-SHA256 audit chain. What we ship.
AAR (Agent Action Receipt): per-action Ed25519 signing. Designed by @Cyberweasel777.
SCC (Session Continuity Certificate): session-level identity with Merkle memory roots, capability hash lineage, and prior-session chaining. Co-designed by @botbotfromuk and @Cyberweasel777 in a public FINOS thread.

AIR Blackbox v1.12.0 detects all three. The goal is not to push our scheme, it is to give your code a clean pass if you have adopted any industry-recognized identity binding.

What a failing scan looks like

If your code is an autonomous agent and none of these schemes are in use, the scanner reports:

Article 12 — Record-Keeping
  FAIL  Agent identity binding
        Autonomous agent detected in 3 file(s) (agent.py, tick.py, loop.py)
        but no stable cryptographic identity binding found.
        Checked for: air-trust, AAR, SCC.

The simplest fix using air-trust

Persist a stable signing key across restarts so every tick is provably signed by the same agent:

from pathlib import Path
import air_trust

KEY_PATH = Path.home() / ".air-trust" / "keys" / "my-agent-ed25519.key"

identity = air_trust.AgentIdentity(
    agent_name="my-agent",
    owner="team@company.com",
    agent_version="1.0.0",
)

def tick():
    with air_trust.trust(identity):
        make_decision()

If you would rather use AAR or SCC, the scanner still passes as long as the library is imported and the key path is persistent. Pick what fits your stack.

Question 2: Will it behave the same way tomorrow?

Earlier this month, Atherik was acquired. They solve this problem at runtime: AI models give different outputs on different GPUs, driver versions, and cuDNN settings. The acquisition is the market confirming the gap is real.

Runtime solutions help if you can afford them. Most teams cannot. And for SR 11-7 model validation, the Federal Reserve guidance governing model risk management in US financial services, reproducibility is a mandatory audit requirement, not an optional feature.

What reproducibility failure looks like

Same model, same seed, same input, on two different GPUs:

import torch

model = SomeModel()
tensor = torch.randn(10, 10).cuda()
output = model(tensor)

Run this on an NVIDIA A100 and on an NVIDIA H100 with identical PyTorch 2.4 and cuDNN 8.9. The two outputs will differ. cuDNN picks different kernels based on hardware capabilities. The model is no longer deterministic. That is an EU AI Act Article 15 robustness violation and an SR 11-7 validation failure.

What the scanner checks

AIR Blackbox v1.12.0 added three checks for this. To my knowledge, these are the first of their kind in compliance tooling. If you know of another open-source tool doing this, please tell me, I want to credit it and learn from it.

Check 1: RNG seeds across all sources

The scanner looks for seed setting across Python's random, NumPy, PyTorch CPU, PyTorch CUDA, TensorFlow, and JAX. Missing any one breaks reproducibility. Partial coverage warns, missing all of them in an ML codebase fails.

Check 2: Deterministic algorithm flags

Seeds alone are not enough. cuDNN defaults to picking the fastest kernel, which is often non-deterministic. TensorFlow ops behave the same way. The scanner looks for these flags in your Python code and in your .env, Dockerfile, YAML, and shell scripts:

import os
import torch

torch.use_deterministic_algorithms(True)
torch.backends.cudnn.deterministic = True
torch.backends.cudnn.benchmark = False
os.environ['CUBLAS_WORKSPACE_CONFIG'] = ':4096:8'

TensorFlow equivalent:

import os
import tensorflow as tf

tf.config.experimental.enable_op_determinism()
os.environ['TF_DETERMINISTIC_OPS'] = '1'

Check 3: Hardware abstraction

Hardcoded .to("cuda") without a capability fallback crashes on CPU-only, Apple Silicon, or AMD hardware. Worse, it silently produces different outputs when you migrate between GPU generations.

Flagged as non-compliant:

model = SomeModel().to("cuda")

Compliant:

import torch

device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
model = SomeModel().to(device)

None of these checks require GPU hardware to run. The scanner reads your code, flags the anti-patterns, and you fix them at CI/CD time before a regulator, auditor, or on-call engineer finds them in production.

Question 3: Can the user understand why it made this decision?

Article 13 of the EU AI Act covers transparency. Six sub-requirements:

13(2): instructions for use
13(3)(a): identity of the provider
13(3)(b): characteristics, capabilities, and limitations
13(3)(c): changes to the system after conformity assessment
13(3)(d): human oversight measures including output interpretation
Article 50: users must be informed they are interacting with AI

Most compliance tools skip Article 13 because it is documentation-heavy and hard to automate. AIR Blackbox v1.12.0 ships what I believe is the first Article 13 static scanner. Again, if I am wrong, please tell me so I can credit the prior work.

What the Article 13 scan looks like

Article 13 — Transparency and Provision of Information
  PASS   AI disclosure to users
  FAIL   Capability and limitation documentation
         No MODEL_CARD.md, SYSTEM_CARD.md, or capability docs found
  WARN   Instructions for use
         Only README.md found. Article 13(2) expects dedicated instructions
  PASS   Provider identity disclosure
  WARN   Output interpretation support
         No confidence scores, rationale, or explanation patterns detected
  PASS   Change logging and versioning

Each check maps to a specific clause. Each failure comes with a fix hint. The output interpretation check catches agents that return decisions without any reasoning trace, confidence score, or rationale. In my experience this is the most common gap in current Python AI codebases.

A concrete example

A Regulation B compliant credit decision needs to be explainable to the applicant. An agent returning a boolean approved / denied without rationale fails this.

Flagged version:

def predict_creditworthiness(applicant):
    return model.predict(applicant)

Passing Article 13(3)(d):

def predict_creditworthiness(applicant):
    prediction = model.predict(applicant)
    confidence = model.predict_proba(applicant).max()
    reasoning = generate_reasoning_trace(applicant, prediction)
    return {
        "decision": prediction,
        "confidence_score": confidence,
        "rationale": reasoning,
    }

The pattern holds across every industry

Every industry where AI makes consequential decisions follows the same structure:

A traditionally analog market goes digital.
AI gets embedded in the decision-making layer.
Those AI decisions fall under EU AI Act Annex III or Colorado SB 205.
Nobody audits the AI layer for compliance.
The regulatory deadline is months away, not years.

My previous post walked through tokenized real estate. 1.4 trillion dollar projected 2026 market. 80 plus platforms globally. Zero of them scanning their AI systems for compliance. The pattern holds everywhere:

Healthcare: AI-assisted clinical decisions are Annex III high-risk. Most clinical AI systems have no agent identity binding, no determinism flags, no structured output interpretation.
Financial services: credit underwriting, trading signals, fraud detection. SR 11-7 requires reproducibility. The hardware determinism check alone flags violations in most codebases.
Insurance: underwriting, claims, risk scoring. Annex III high-risk. Article 13 transparency obligations apply directly.
Hiring and HR: resume screening, interview scoring. Explicitly listed in Annex III. Agent identity continuity matters because hiring decisions affect protected classes.

The opportunity is not any single industry. It is the open-source compliance infrastructure underneath all of them, and that infrastructure is being built in public, by people like the ones I credited above, right now.

What this tool is not

This part matters. A good compliance scanner cannot be the only thing standing between your AI agent and a regulator.

This checks technical requirements, not legal compliance. It is a linter, not a lawyer.
Passing every check does not mean you are legally compliant. It means your code implements the technical controls the regulation references.
Legal interpretation is your counsel's job.
Static analysis cannot catch every runtime issue. Pair it with trust-layer integrations for runtime evidence.
Pattern-based detection has false positives. If you see one, report it, it makes the scanner better.

How to help

If you read this far, you probably work on AI systems that will need to pass an audit. The scanner only gets better with your feedback.

Try it. pip install air-blackbox && air-blackbox comply --scan .
If it misses a pattern, open an issue. Include the code pattern and the article it should map to. Every issue is a new check.
If it produces a false positive on your code, open an issue. Every correction flows into training data for a fine-tuned compliance model, so false positives become easier to catch next time.
If you are building in the same space (identity, determinism, audit trails), I would like to coordinate. Ping me on the repo or on @jshotwell.

Try it

pip install air-blackbox
cd /path/to/your/ai/project
air-blackbox comply --scan . -v

Useful links

Repo: github.com/airblackbox/air-trust
PyPI: pypi.org/project/air-blackbox
Docs and demos: airblackbox.ai
Interactive browser demo: airblackbox.ai/demo/hub

Thanks

@botbotfromuk and @Cyberweasel777 for the SCC and AAR specs that v1.12.0 detects.
The FINOS AI Governance Framework maintainers, whose public thread on NIST RFI Docket NIST-2025-0035 shaped a big part of this release.
Everyone who has filed an issue on the scanner. Every correction is a data point.

109 days to August 2, 2026. Run the scan. See what is missing. Fix it before someone else's audit does it under pressure.

The Unaudited AI Layer: Why Every Industry Running AI Transactions Needs a Compliance Check

Jason Shotwell — Sun, 12 Apr 2026 23:10:55 +0000

Every major industry is quietly embedding AI into its transaction layer. Property valuations. Insurance underwriting. Lending decisions. Medical diagnostics. Hiring algorithms. These AI systems make millions of consequential decisions per day, and almost none of them are being audited for regulatory compliance.

The EU AI Act goes into enforcement on August 2, 2026. Colorado's AI Act follows close behind. And the AI systems making your industry's highest-stakes decisions? Most of them have never been scanned for compliance with the regulations that now govern them.

I built AIR Blackbox, an open-source CLI tool that scans Python AI projects for EU AI Act technical requirements. It checks six articles covering risk management, data governance, documentation, record-keeping, human oversight, and robustness. One install, one scan, one report.

But here is what I have been thinking about: the same compliance gap exists in every industry where AI makes decisions that affect people's lives, money, or access to services. Let me walk through where this matters most.

The Pattern

Every industry below follows the same structure:

A massive, traditionally analog market goes digital
AI gets embedded into the decision-making layer
Those AI decisions fall under high-risk classification in the EU AI Act, Colorado SB 205, or both
Nobody is auditing the AI layer for compliance
The regulatory deadline is months away

The opportunity is not in any single industry. It is in the compliance infrastructure that sits underneath all of them.

1. Tokenized Real Estate and Real-World Assets ($1.4T projected 2026)

Tokenization platforms use AI for property valuation, investor risk scoring, automated KYC/AML, and fraud detection. Every one of those AI systems makes "consequential decisions" about people's investments.

The compliance gap: platforms spend 30% of their budget on securities and blockchain compliance. They spend 0% on auditing the AI systems that actually make the decisions.

Regulatory exposure: EU AI Act (essential financial services, Annex III), MiCA (crypto-asset service providers), Colorado SB 205 (consequential financial decisions), SEC (securities compliance).

There are 80+ tokenization service providers globally. Zero of them have AI governance tooling.

2. Insurance and InsurTech

AI underwriting models decide who gets coverage and at what price. Claims processing AI determines whether your claim gets paid or denied. Fraud detection AI flags (or misses) suspicious activity.

These are textbook high-risk AI systems under the EU AI Act. Insurance is explicitly called out in Annex III as an essential service where AI decisions require full compliance.

The compliance gap: InsurTech companies have built sophisticated ML models for pricing and claims, but the governance layer (documentation, audit trails, human oversight, bias detection) is often bolted on as an afterthought, if it exists at all.

Market size: The global InsurTech market is projected to exceed $150B by 2030. Every AI-driven underwriting model in Europe needs to satisfy Articles 9-15 by August 2026.

3. Lending and Credit Decisioning (FinTech)

AI credit scoring determines who gets a loan, what interest rate they pay, and whether they get approved or denied. This is one of the most heavily regulated AI use cases on the planet.

The compliance gap: traditional banks have compliance departments. FinTech startups building AI lending products often don't. They have ML engineers building scoring models and growth teams optimizing approval rates, but the Article 12 audit trail? The Article 14 human override? Often missing.

Regulatory exposure: EU AI Act (credit and financial services, Annex III), Colorado SB 205 (consequential credit decisions), Fair Lending laws (US), Consumer Duty (UK).

4. Healthcare AI and MedTech

Diagnostic AI (radiology, pathology, dermatology), treatment recommendation engines, clinical decision support, mental health chatbots, and drug interaction checkers. Every one of these makes decisions that directly affect patient outcomes.

The compliance gap: the FDA has a pathway for AI/ML-based Software as a Medical Device (SaMD). But FDA clearance does not cover EU AI Act compliance. A diagnostic AI can be FDA-cleared AND non-compliant with the EU AI Act simultaneously. Two separate compliance requirements, two separate audit processes, and most companies are only thinking about one of them.

Regulatory exposure: EU AI Act (safety component in medical devices, Annex I + Annex III for health-related AI), MDR (Medical Devices Regulation), FDA (US), state-level healthcare AI transparency laws.

5. HR Tech and Recruitment AI

Resume screening AI, candidate scoring models, automated interview analysis, workforce analytics, and performance management AI. Employment is one of the most explicitly regulated AI categories globally.

This is one of the first verticals where enforcement has already started. New York City's Local Law 144 requires bias audits for automated employment decision tools. The EU AI Act classifies employment AI as high-risk. Colorado SB 205 covers AI making employment decisions.

The compliance gap: many HR Tech vendors market "AI-powered hiring" without the infrastructure to prove their models are free of bias, properly documented, or equipped with human oversight. The marketing moved faster than the governance.

6. Autonomous Vehicles and Mobility

Self-driving vehicle AI, fleet management optimization, route planning, driver safety scoring, and predictive maintenance. The AI is making life-safety decisions at highway speed.

The compliance gap: automotive OEMs have strong safety testing cultures. But the EU AI Act introduces requirements beyond safety testing: documentation standards, audit trails, and transparency obligations that traditional automotive compliance processes were not designed to cover.

Regulatory exposure: EU AI Act (safety component in vehicles, Annex I), existing vehicle type-approval regulations, emerging UNECE regulations for autonomous driving.

7. EdTech and AI Tutoring

Adaptive learning platforms, automated grading, student performance prediction, dropout risk scoring, and AI-generated educational content. Education is explicitly listed in the EU AI Act's high-risk categories.

The compliance gap: EdTech companies have moved aggressively into AI-powered personalization, but few have the documentation, bias detection, or human oversight mechanisms that the EU AI Act requires. A student scoring model that determines course placement is a high-risk AI system, whether the company building it realizes that or not.

8. Legal Tech

Contract analysis AI, legal research assistants, case outcome prediction, document review automation, and AI-generated legal briefs. Ironic that the tools lawyers use may themselves be non-compliant.

The compliance gap: legal AI tools process privileged information and influence case strategy. The EU AI Act classifies AI used in the administration of justice as high-risk. Most legal AI vendors focus on accuracy and speed, not on the governance infrastructure that regulators will demand.

The Common Thread

Every industry above has the same profile:

AI is making decisions that materially affect people
Regulations now classify those AI systems as high-risk
The compliance infrastructure (documentation, audit trails, human oversight, bias detection, robustness testing) is either incomplete or absent
The enforcement deadline is months away
Nobody is scanning the AI layer

What AIR Blackbox Does

AIR Blackbox is an open-source CLI tool that scans Python AI projects for compliance with six EU AI Act technical requirements:

pip install air-blackbox
air-blackbox scan .

It checks:

Article 9: Risk management system
Article 10: Data governance
Article 11: Technical documentation
Article 12: Record-keeping and audit trails
Article 14: Human oversight
Article 15: Accuracy, robustness, cybersecurity

With Phase 2 (shipping now), it maps results across the EU AI Act, ISO 42001, NIST AI RMF, and Colorado SB 205 simultaneously. One scan, four frameworks.

The scanner does not care what industry you are in. It cares whether your AI system has the technical controls that regulators require. A property valuation model and a credit scoring model need the same six compliance checks. The regulations are the same. The scan is the same.

The Bigger Vision

I think of AIR Blackbox as the compliance verification layer for AI transactions. The same way a financial audit verifies that accounting standards are met, an AIR Blackbox scan verifies that AI governance standards are met.

Every tokenized real estate transaction should carry an AIR Blackbox evidence bundle proving the valuation AI was audited. Every AI lending decision should reference a signed compliance report. Every autonomous vehicle software update should pass a governance scan before deployment.

The industries are different. The compliance requirement is the same. And right now, with 4 months until the EU AI Act's August 2026 deadline, most of these industries are flying blind.

Try It

AIR Blackbox is open-source and free:

pip install air-blackbox
air-blackbox scan your-project/

GitHub: github.com/air-blackbox
Website: airblackbox.ai

If you are building AI systems in any of the industries above, scan your project. The results might surprise you.

Disclaimer: AIR Blackbox scans for technical requirements. This is not a certified compliance test. It is a starting point to identify potential gaps. Consult a qualified attorney for legal compliance guidance.

I'm Jason Shotwell, the builder behind AIR Blackbox. I write about AI governance, open-source compliance tooling, and the race to August 2026. Follow me on Dev.to for more.

Cryptographic Proof of Agent-to-Agent Handoffs in Python

Jason Shotwell — Sat, 11 Apr 2026 04:24:54 +0000

When your AI pipeline hands off from one agent to another, how do you prove it happened?

Not "there's a log entry." Prove it. Cryptographically. In a way that holds up when someone asks: which agent made this decision, and did it actually receive the data it claimed to receive?

That's what I shipped this week in air-trust v0.6.1: Ed25519 signed handoffs for multi-agent Python systems.

The Problem

Multi-agent pipelines are becoming the default architecture for serious AI work. A research agent gathers context, hands off to a writer agent, which hands off to a fact-checker, which hands off to a publisher. Each agent does a piece of the work.

But when something goes wrong — or when a regulator asks for an audit trail — you have a problem. Your logs show what happened. They don't prove who did it or that the data wasn't modified in transit.

Three specific failure modes that are genuinely hard to detect today:

Payload tampering: Agent A says it handed off document X. Agent B received document X. But was it the same X? Without a hash comparison locked in at handoff time, you can't know.
Identity spoofing: An agent claims to be "research-bot." Is it? If agents communicate over any shared message bus, impersonation is trivial.
Silent unsigned records: You think your audit chain has signatures. It doesn't — the signing key was missing and the library failed silently. (This was actually a bug in air-trust v0.6.0 that we fixed in v0.6.1.)

The EU AI Act's Article 12 requires high-risk AI systems to maintain logs "sufficient to ensure traceability." Unsigned JSON files don't meet that bar for systems making consequential decisions.

The Solution: Three Record Types + Ed25519 Keys

air-trust adds three new event types to its audit chain:

handoff_request — Agent A says "I want to hand off to Agent B with this payload"
handoff_ack — Agent B acknowledges receipt
handoff_result — Agent B reports back what it produced

Each record is automatically signed with the agent's Ed25519 private key. The verifier checks all three: valid signatures, matching counterparties, payload hash integrity, and nonce uniqueness (to prevent replay attacks).

Install

pip install "air-trust[handoffs]"

The [handoffs] extra pulls in the cryptography library for Ed25519 support. The core audit chain has no external dependencies.

Generate keypairs for each agent

python3 -m air_trust keygen --agent research-bot
python3 -m air_trust keygen --agent writer-bot

Keys are stored at ~/.air-trust/keys/ with 0600 permissions.

Instrument the handoff

import air_trust
from air_trust import trust, session, AuditChain

# Research agent side
chain = AuditChain()
air_trust.trust(identity=air_trust.AgentIdentity(
    agent_id="research-bot",
    fingerprint="research-bot"
))

with session(chain):
    # ... do research work ...

    iid = chain.handoff_request(
        counterparty_id="writer-bot",
        payload={"summary": research_summary},
    )

# Writer agent side (separate process, same or different machine)
air_trust.trust(identity=air_trust.AgentIdentity(
    agent_id="writer-bot",
    fingerprint="writer-bot"
))

with session(chain):
    chain.handoff_ack(
        interaction_id=iid,
        counterparty_id="research-bot",
    )

    # ... do writing work ...

    chain.handoff_result(
        interaction_id=iid,
        counterparty_id="research-bot",
        payload={"article": finished_article},
    )

Verify the chain

python3 -m air_trust verify audit_chain.jsonl

Output:

INTEGRITY     PASS  47 events, 47 valid HMAC links
COMPLETENESS  PASS  2 sessions complete, no gaps
HANDOFFS      PASS  1 interaction verified

  interaction abc123:
    request   PASS  Ed25519 OK (research-bot)
    ack       PASS  Ed25519 OK (writer-bot)
    result    PASS  Ed25519 OK (writer-bot)
    payload   PASS  SHA-256 hash match
    nonce     PASS  unique

If someone tampers with the payload between request and result:

HANDOFFS      FAIL  1 interaction failed

  interaction abc123:
    result    FAIL  payload hash mismatch
              expected: a3f2c1...
              got:      9d4e8b...

How It Works Under the Hood

The signing payload

Every signed record commits to six fields, pipe-delimited:

interaction_id|counterparty_id|payload_hash|nonce|type|timestamp

The payload_hash is SHA-256 of the JSON-serialized payload. This means:

The signature covers the payload content, not just the record metadata
Payload mutation is detectable even if the signature itself isn't forged
The nonce prevents an attacker from replaying a valid old record

Ed25519 vs. HMAC

The tamper-evident chain (spec v1.0) uses HMAC-SHA256 with a shared secret — this catches post-hoc modification of stored records. But HMAC is symmetric: anyone with the secret key can forge a record.

Signed handoffs (spec v1.2) use Ed25519 asymmetric keys. The private key never leaves the agent. The public key is embedded in every signed record. This means:

Non-repudiation: agent A can prove it signed the request, and nobody else can produce that signature
No shared secret: agents don't need to trust each other's key management
Public key in the record: verifiers don't need a key registry — the public key is self-contained in the audit log

The audit chain is layered

v1.0  HMAC chain        — tamper detection for all records
v1.1  Session sequences — completeness: no gaps, no replays within a session
v1.2  Signed handoffs   — identity proof at agent boundaries

Each layer is backward compatible. A v1.0 chain verifies clean under v1.2. A v1.2 chain with no handoffs still passes.

What We Fixed in v0.6.1

The silent failure modes I mentioned earlier were real bugs in v0.6.0, caught during a design review after shipping:

Bug 1 — Signed records written without signatures: If the cryptography package wasn't installed, handoff records were written silently without signatures. The verifier then silently skipped them. The chain appeared to verify clean. It now raises ImportError at write time instead.

Bug 2 — Verifier skipped unsigned records: Even with the library installed, if an agent had no keypair, records were written unsigned and the verifier skipped checking them. It now flags these as missing_signature with severity warn.

Bug 3 — Session ID leaked on exception: If a session() block raised, the ContextVar holding the session ID wasn't reset, so the next session wrote events with the wrong session ID. Fixed with a finally block.

Bug 4 — Thread safety: The global chain and identity singletons weren't protected by a lock. In threaded code (common with agent frameworks), you could get cross-thread identity clobbering. Fixed with threading.Lock().

I'm documenting these because they're the kind of bugs that would be catastrophic in a compliance context — you think you have signed records, you don't. Soundness matters more than features here.

What This Is (and Isn't)

This is a technical audit layer, not a legal compliance certification. air-trust helps you answer: did these agents interact in the way the logs claim, and is the data intact? It doesn't answer whether your system meets every requirement of the EU AI Act — that's a legal question involving risk classification, conformity assessments, and a lot more.

Think of it as a linter for your audit chain. Fast, local, no cloud, no API keys. It either passes or it tells you exactly what's wrong.

Try It

pip install "air-trust[handoffs]"
python3 -m air_trust keygen --agent my-agent
python3 examples/signed_handoff.py  # in the repo
python3 -m air_trust verify audit_chain.jsonl

Interactive demo (no install): airblackbox.ai/demo/signed-handoff

GitHub: github.com/airblackbox/air-trust

What's Next

The roadmap has two items queued for Phase 3:

Remote verification endpoint — post a chain to a verifier service and get a signed attestation back (useful for third-party audits)
Framework trust layers — drop-in air_trust wrappers for LangChain, CrewAI, and AutoGen that auto-instrument handoffs with zero code changes

If you're building multi-agent systems and care about auditability, I'd genuinely like to know what your current setup looks like. Comments open.

Write AI Policies That Actually Work: Custom Rule Examples

Jason Shotwell — Fri, 03 Apr 2026 14:46:01 +0000

Write AI Policies That Actually Work: Custom Rule Examples

Most AI governance policies are as useful as a chocolate teapot — they look impressive in meetings but melt the moment they touch production heat.

The Problem: Your AI Policies Are Theater, Not Engineering

Here's what I see in every "AI governance" document that crosses my desk:

"AI systems must be fair and unbiased" (What does that mean? Fair to whom? Measured how?)
"Models must be regularly monitored" (How often is regular? What metrics matter?)
"Data privacy must be maintained" (Which data? What constitutes a violation?)

These aren't policies. They're wishes written in corporate speak.

Meanwhile, your AI agents are burning through API quotas, hallucinating customer data, and making decisions that would make a compliance officer weep. You need rules that actually fire when something goes wrong, not mission statements that make executives feel better about their AI initiatives.

The gap between "we have AI governance" and "our AI governance actually works" is filled with custom rules that trigger on specific, measurable violations. Not vibes. Not best practices. Executable code that says "this specific thing happened, therefore that specific action must be taken."

Architecture: How Policy Rules Actually Work

Here's how Airblackbox turns your governance requirements into executable policies:

graph TD
    A[AI Agent Request] --> B[Gateway Proxy]
    B --> C[LLM Provider]
    C --> D[Response Capture]
    D --> E[Rule Engine]
    E --> F{Policy Check}
    F -->|Pass| G[Log & Forward]
    F -->|Fail| H[Block & Alert]
    F -->|Warn| I[Flag & Forward]

    J[Custom Rules] --> E
    K[EU AI Act Rules] --> E
    L[Rate Limits] --> E

    H --> M[Compliance Dashboard]
    I --> M
    G --> N[Observability Store]

    style F fill:#ff6b6b
    style J fill:#4ecdc4
    style M fill:#45b7d1

The Gateway sits between your agent and the LLM, captures everything, runs your custom rules against the traffic, and either blocks, warns, or passes through based on what it finds.

Think of it as a firewall, but instead of blocking ports, it blocks prompts that violate your policies.

Implementation: Building Custom Policy Rules

Let's build three real rules that solve actual problems developers face.

Rule 1: Rate Limiting by User Role

Problem: Your intern shouldn't be able to burn through the entire monthly GPT-4 budget in one afternoon experimenting with creative writing prompts.

# custom_rules/rate_limiting.py
from airblackbox.rules import Rule, RuleResult
from datetime import datetime, timedelta
import redis

class UserRoleLimiter(Rule):
    def __init__(self):
        self.redis = redis.Redis(host='localhost', port=6379, db=0)
        self.limits = {
            'intern': {'requests_per_hour': 10, 'tokens_per_day': 1000},
            'developer': {'requests_per_hour': 50, 'tokens_per_day': 10000},
            'admin': {'requests_per_hour': 200, 'tokens_per_day': 50000}
        }

    def evaluate(self, request, response, metadata):
        user_id = metadata.get('user_id')
        user_role = metadata.get('user_role', 'intern')  # Default to most restrictive

        if user_role not in self.limits:
            return RuleResult(
                passed=False,
                message=f"Unknown user role: {user_role}",
                action="block"
            )

        # Check hourly request limit
        hour_key = f"requests:{user_id}:{datetime.now().strftime('%Y%m%d%H')}"
        hourly_requests = self.redis.incr(hour_key)
        self.redis.expire(hour_key, 3600)  # Expire after 1 hour

        if hourly_requests > self.limits[user_role]['requests_per_hour']:
            return RuleResult(
                passed=False,
                message=f"User {user_id} exceeded hourly limit ({self.limits[user_role]['requests_per_hour']})",
                action="block",
                metadata={'current_requests': hourly_requests}
            )

        # Check daily token limit
        day_key = f"tokens:{user_id}:{datetime.now().strftime('%Y%m%d')}"
        current_tokens = int(self.redis.get(day_key) or 0)
        estimated_tokens = len(request.get('prompt', '')) // 4  # Rough estimation

        if current_tokens + estimated_tokens > self.limits[user_role]['tokens_per_day']:
            return RuleResult(
                passed=False,
                message=f"User {user_id} would exceed daily token limit",
                action="block",
                metadata={'current_tokens': current_tokens, 'estimated_tokens': estimated_tokens}
            )

        # Update token counter after successful validation
        self.redis.incrby(day_key, estimated_tokens)
        self.redis.expire(day_key, 86400)  # Expire after 24 hours

        return RuleResult(
            passed=True,
            message=f"Rate limit check passed for {user_role}",
            metadata={'requests_remaining': self.limits[user_role]['requests_per_hour'] - hourly_requests}
        )

Rule 2: PII Detection and Redaction

Problem: Your customer service agent just tried to send someone's SSN to OpenAI. This is not ideal for anyone involved.

# custom_rules/pii_protection.py
import re
from airblackbox.rules import Rule, RuleResult

class PIIProtectionRule(Rule):
    def __init__(self):
        self.patterns = {
            'ssn': r'\b\d{3}-?\d{2}-?\d{4}\b',
            'credit_card': r'\b\d{4}[- ]?\d{4}[- ]?\d{4}[- ]?\d{4}\b',
            'email': r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b',
            'phone': r'\b\d{3}[-.]?\d{3}[-.]?\d{4}\b',
            'ip_address': r'\b(?:\d{1,3}\.){3}\d{1,3}\b'
        }

    def evaluate(self, request, response, metadata):
        prompt = request.get('prompt', '')
        violations = []
        redacted_prompt = prompt

        for pii_type, pattern in self.patterns.items():
            matches = re.finditer(pattern, prompt, re.IGNORECASE)
            for match in matches:
                violations.append({
                    'type': pii_type,
                    'value': match.group(),
                    'position': match.span()
                })
                # Redact the PII
                redacted_prompt = redacted_prompt.replace(
                    match.group(), 
                    f'[REDACTED_{pii_type.upper()}]'
                )

        if violations:
            # For high-risk PII, block entirely
            high_risk = ['ssn', 'credit_card']
            if any(v['type'] in high_risk for v in violations):
                return RuleResult(
                    passed=False,
                    message=f"High-risk PII detected: {[v['type'] for v in violations]}",
                    action="block",
                    metadata={'violations': violations}
                )

            # For lower-risk PII, redact and warn
            return RuleResult(
                passed=True,
                message=f"PII detected and redacted: {[v['type'] for v in violations]}",
                action="modify",
                metadata={
                    'violations': violations,
                    'modified_prompt': redacted_prompt
                }
            )

        return RuleResult(passed=True, message="No PII detected")

Rule 3: Content Appropriateness with Business Context

Problem: Your legal research agent keeps trying to get ChatGPT to write creative fiction about murder trials. Your lawyers are not amused.

# custom_rules/content_appropriateness.py
from airblackbox.rules import Rule, RuleResult
import openai

class ContentAppropriatenessRule(Rule):
    def __init__(self, allowed_domains=None):
        self.client = openai.OpenAI()  # For moderation API
        self.allowed_domains = allowed_domains or []

        # Business context keywords that make otherwise flagged content acceptable
        self.business_contexts = {
            'legal_research': ['case law', 'legal precedent', 'court ruling', 'litigation'],
            'security_research': ['vulnerability', 'penetration test', 'security audit'],
            'medical_research': ['clinical trial', 'medical study', 'patient care'],
            'content_moderation': ['content policy', 'moderation guidelines', 'user safety']
        }

    def evaluate(self, request, response, metadata):
        prompt = request.get('prompt', '')
        user_domain = metadata.get('user_domain', 'general')

        # Run OpenAI's moderation check
        try:
            moderation_response = self.client.moderations.create(input=prompt)
            flagged = moderation_response.results[0].flagged
            categories = moderation_response.results[0].categories
        except Exception as e:
            return RuleResult(
                passed=False,
                message=f"Moderation check failed: {str(e)}",
                action="block"
            )

        if not flagged:
            return RuleResult(passed=True, message="Content passed moderation")

        # Content is flagged, check for business context exceptions
        flagged_categories = [cat for cat, flagged in categories.__dict__.items() if flagged]

        # Check if user domain allows this type of content
        if user_domain in self.business_contexts:
            context_keywords = self.business_contexts[user_domain]
            if any(keyword.lower() in prompt.lower() for keyword in context_keywords):
                return RuleResult(
                    passed=True,
                    message=f"Content flagged but allowed for {user_domain} context",
                    action="warn",
                    metadata={
                        'flagged_categories': flagged_categories,
                        'business_context': user_domain,
                        'justification': 'Business context exception applied'
                    }
                )

        # No business context exception, block the request
        return RuleResult(
            passed=False,
            message=f"Content flagged for: {', '.join(flagged_categories)}",
            action="block",
            metadata={'flagged_categories': flagged_categories}
        )

Wiring It All Together

Now let's create a policy engine that runs all these rules:

# policy_engine.py
from airblackbox import Gateway
from custom_rules.rate_limiting import UserRoleLimiter
from custom_rules.pii_protection import PIIProtectionRule
from custom_rules.content_appropriateness import ContentAppropriatenessRule

# Initialize the gateway with custom rules
gateway = Gateway()

# Add your custom rules
gateway.add_rule(UserRoleLimiter())
gateway.add_rule(PIIProtectionRule())
gateway.add_rule(ContentAppropriatenessRule(
    allowed_domains=['legal_research', 'security_research']
))

# Start the gateway
if __name__ == "__main__":
    gateway.start(host="0.0.0.0", port=8080)

Pitfalls: What Will Break and How to Fix It

1. Rule Ordering Matters

Problem: Your PII redaction rule runs after your rate limiting rule, so you're counting tokens for text that gets modified.

Solution: Rules run in the order you add them. Put modification rules (like PII redaction) first, then validation rules (like rate limits).

# Wrong order
gateway.add_rule(UserRoleLimiter())  # Counts original tokens
gateway.add_rule(PIIProtectionRule())  # Then redacts

# Right order  
gateway.add_rule(PIIProtectionRule())  # Redacts first
gateway.add_rule(UserRoleLimiter())  # Counts redacted tokens

2. Performance Death by A Thousand Cuts

Problem: You added 20 rules that each make external API calls. Your response time went from 200ms to 5 seconds.

Solution: Cache expensive operations and use async where possible:

from functools import lru_cache
import asyncio

class OptimizedPIIRule(Rule):
    @lru_cache(maxsize=1000)
    def _check_patterns(self, text_hash, text):
        # Expensive regex operations cached by hash
        return self._find_pii_violations(text)

    async def evaluate_async(self, request, response, metadata):
        # Run expensive operations in parallel
        text_hash = hash(request.get('prompt', ''))
        violations = await asyncio.to_thread(
            self._check_patterns, text_hash, request.get('prompt', '')
        )
        return self._build_result(violations)

3. State Management in Distributed Systems

Problem: You're running multiple gateway instances behind a load balancer. Rate limits work inconsistently because each instance has its own Redis connection and state.

Solution: Use Redis Cluster or a shared state backend:

import redis.sentinel

class DistributedUserRoleLimiter(Rule):
    def __init__(self):
        # Use Redis Sentinel for high availability
        sentinel = redis.sentinel.Sentinel([
            ('localhost', 26379),
            ('localhost', 26380),
            ('localhost', 26381)
        ])
        self.redis = sentinel.master_for('mymaster', socket_timeout=0.1)

Measurement: How to Know It's Working

Your rules are only as good as your ability to measure their effectiveness. Here's how to build observability into your policy engine:

# monitoring.py
from dataclasses import dataclass
from datetime import datetime
import json

@dataclass
class RuleMetrics:
    rule_name: str
    executions: int
    blocks: int
    warnings: int
    avg_execution_time: float
    last_violation: datetime

class PolicyMonitor:
    def __init__(self):
        self.metrics = {}

    def record_execution(self, rule_name, execution_time, result):
        if rule_name not in self.metrics:
            self.metrics[rule_name] = {
                'executions': 0, 'blocks': 0, 'warnings': 0,
                'total_time': 0, 'last_violation': None
            }

        m = self.metrics[rule_name]
        m['executions'] += 1
        m['total_time'] += execution_time

        if result.action == 'block':
            m['blocks'] += 1
            m['last_violation'] = datetime.now()
        elif result.action == 'warn':
            m['warnings'] += 1

    def get_dashboard_data(self):
        dashboard = {}
        for rule_name, data in self.metrics.items():
            dashboard[rule_name] = RuleMetrics(
                rule_name=rule_name,
                executions=data['executions'],
                blocks=data['blocks'],
                warnings=data['warnings'],
                avg_execution_time=data['total_time'] / max(data['executions'], 1),
                last_violation=data['last_violation']
            )
        return dashboard

Key metrics to track:

Rule execution frequency (are your rules actually running?)
Block/warn rates (too high = rules too strict, too low = rules not catching issues)
False positive rates (measure via manual review of blocked requests)
Performance impact (rule execution time vs total request time)

Next Steps

You now have the blueprint for AI policies that actually enforce themselves. But reading about rules and running them in production are different beasts entirely.

Want to see this in action? Clone the Airblackbox policy examples repo and run these rules against real traffic. The repo includes:

Complete working examples of all three rules above
Docker Compose setup with Redis and monitoring dashboards
Test scenarios that trigger each rule type
Performance benchmarks for rule execution

Or jump straight into the deep end: Install Airblackbox, point it at your existing AI agents, and watch what your policies actually catch. You'll be surprised what your agents are trying to do when you're not looking.

Because the best AI governance policy is the one that runs itself. Everything else is just expensive theater.

Try Airblackbox — Because your AI agents need adult supervision, not mission statements.

Why Your Streaming AI Agent Looks Broken (And How to Fix It)

Jason Shotwell — Thu, 02 Apr 2026 14:59:23 +0000

Why Your Streaming AI Agent Looks Broken (And How to Fix It)

Your streaming AI agent appears to think for 30 seconds, then vomits a wall of text all at once — congratulations, you've built a very expensive typewriter with performance anxiety.

The Problem: When "Streaming" Isn't Actually Streaming

You've hooked up your beautiful AI agent to OpenAI's streaming API. The docs promise smooth, real-time token delivery. Your code looks perfect. But users are staring at loading spinners for eons, then getting hit with text dumps that would make a fire hose jealous.

The culprit? Gateway buffering.

Every reverse proxy, load balancer, and observability tool between your agent and OpenAI is helpfully "optimizing" your stream by collecting tokens into neat little batches. Your streaming response gets turned into a buffered response, and your users get a front-row seat to watching paint dry.

This isn't just a UX problem — it's an architecture problem. When your AI agent's thinking process is invisible, users assume it's broken. When responses arrive in chunks instead of flowing naturally, the illusion of intelligence shatters.

The deeper issue: most observability solutions for AI agents weren't designed with streaming in mind. They intercept, process, and forward — which is exactly what you don't want when every millisecond matters for perceived responsiveness.

Architecture: How Streaming Actually Works (When It Works)

graph TD
    A[User Request] --> B[Your AI Agent]
    B --> C[Airblackbox Gateway]
    C --> D[OpenAI API]

    D --> E[Token Stream]
    E --> F[Gateway Processing]
    F --> G[Buffering Decision Point]

    G -->|Bad Path| H[Buffer Tokens]
    H --> I[Batch Forward]
    I --> J[Wall of Text]

    G -->|Good Path| K[Stream Through]
    K --> L[Real-time Tokens]
    L --> M[Smooth User Experience]

    style H fill:#ff9999
    style I fill:#ff9999
    style J fill:#ff9999
    style K fill:#99ff99
    style L fill:#99ff99
    style M fill:#99ff99

The critical insight: observability and streaming are not mutually exclusive. You can record everything that flows through your system without breaking the flow. The gateway needs to be smart enough to tee the stream — capturing data for debugging while preserving the real-time experience.

Implementation: Building a Streaming-First Gateway

Let's build this properly. We'll create a streaming gateway that captures everything for debugging without breaking the user experience.

Step 1: Set Up the Streaming Gateway

import asyncio
import json
import time
from typing import AsyncGenerator, Dict, Any
from fastapi import FastAPI, Request, Response
from fastapi.responses import StreamingResponse
import httpx
import uvicorn

class StreamingGateway:
    def __init__(self, target_base_url: str = "https://api.openai.com"):
        self.target_base_url = target_base_url
        self.client = httpx.AsyncClient()
        self.recorded_calls = []

    async def proxy_stream(self, request: Request) -> StreamingResponse:
        """Proxy streaming requests while capturing data"""

        # Capture request metadata
        call_start = time.time()
        request_body = await request.body()
        request_data = json.loads(request_body) if request_body else {}

        call_record = {
            "id": f"call_{int(time.time() * 1000)}",
            "timestamp": call_start,
            "method": request.method,
            "path": str(request.url.path),
            "request": request_data,
            "response_tokens": [],
            "latency_ms": None,
            "status": "streaming"
        }

        # Forward request to OpenAI
        target_url = f"{self.target_base_url}{request.url.path}"
        headers = dict(request.headers)

        # Remove hop-by-hop headers that break streaming
        headers.pop('host', None)
        headers.pop('content-length', None)

        async def stream_and_record():
            """Stream response while recording tokens"""
            try:
                async with self.client.stream(
                    request.method,
                    target_url,
                    headers=headers,
                    content=request_body,
                    timeout=60.0
                ) as response:

                    # Forward response headers
                    response_headers = dict(response.headers)

                    # Critical: preserve streaming headers
                    if 'transfer-encoding' in response_headers:
                        del response_headers['content-length']

                    # Stream tokens in real-time
                    async for chunk in response.aiter_bytes():
                        if chunk:
                            # Record chunk for debugging (non-blocking)
                            self._record_chunk(call_record, chunk)

                            # IMMEDIATELY yield to user (this is the magic)
                            yield chunk

                    # Finalize recording
                    call_record["latency_ms"] = (time.time() - call_start) * 1000
                    call_record["status"] = "completed"
                    self.recorded_calls.append(call_record)

            except Exception as e:
                call_record["error"] = str(e)
                call_record["status"] = "error"
                self.recorded_calls.append(call_record)
                raise

        return StreamingResponse(
            stream_and_record(),
            media_type="text/plain",
            headers={"Cache-Control": "no-cache"}
        )

    def _record_chunk(self, call_record: Dict[str, Any], chunk: bytes):
        """Record streaming chunk without blocking"""
        try:
            chunk_text = chunk.decode('utf-8')
            if chunk_text.startswith('data: '):
                data_line = chunk_text[6:].strip()
                if data_line != '[DONE]':
                    token_data = json.loads(data_line)
                    if 'choices' in token_data:
                        for choice in token_data['choices']:
                            if 'delta' in choice and 'content' in choice['delta']:
                                call_record["response_tokens"].append({
                                    "content": choice['delta']['content'],
                                    "timestamp": time.time()
                                })
        except:
            # Don't break streaming for recording failures
            pass

# FastAPI app
app = FastAPI(title="Streaming Gateway")
gateway = StreamingGateway()

@app.api_route("/v1/{path:path}", methods=["GET", "POST", "PUT", "DELETE"])
async def proxy_handler(request: Request):
    """Handle all OpenAI API routes"""
    return await gateway.proxy_stream(request)

@app.get("/debug/calls")
async def get_recorded_calls():
    """Debug endpoint to inspect recorded calls"""
    return gateway.recorded_calls

Step 2: Configure Your AI Agent to Use the Gateway

import openai
from typing import AsyncGenerator

class StreamingAgent:
    def __init__(self, gateway_url: str = "http://localhost:8000"):
        # Point OpenAI client to your gateway instead of api.openai.com
        self.client = openai.AsyncOpenAI(
            api_key="your-openai-key",
            base_url=gateway_url + "/v1"  # Gateway intercepts here
        )

    async def stream_response(self, prompt: str) -> AsyncGenerator[str, None]:
        """Stream AI response through the gateway"""

        stream = await self.client.chat.completions.create(
            model="gpt-4",
            messages=[{"role": "user", "content": prompt}],
            stream=True,
            temperature=0.7
        )

        async for chunk in stream:
            if chunk.choices[0].delta.content:
                # This arrives in real-time thanks to the gateway
                yield chunk.choices[0].delta.content

# Usage example
async def demo_streaming():
    agent = StreamingAgent()

    print("Streaming response:")
    async for token in agent.stream_response("Explain quantum computing in simple terms"):
        print(token, end="", flush=True)
    print("\n\nDone!")

if __name__ == "__main__":
    asyncio.run(demo_streaming())

Step 3: Run the Complete System

# Terminal 1: Start the gateway
python streaming_gateway.py
uvicorn streaming_gateway:app --port 8000 --reload

# Terminal 2: Run your agent
python streaming_agent.py

Pitfalls: What Will Break and How to Handle It

1. Header Hell

Problem: HTTP headers like content-length break streaming.

Fix: Strip hop-by-hop headers and preserve transfer-encoding: chunked.

# Bad - keeps content-length
headers = dict(request.headers)

# Good - removes streaming-breaking headers
headers = dict(request.headers)
headers.pop('host', None)
headers.pop('content-length', None)

2. Timeout Disasters

Problem: Default timeouts kill long-running streams.

Fix: Set generous timeouts and handle partial failures gracefully.

# Bad - will timeout on long responses
async with httpx.AsyncClient() as client:
    # Uses default 5-second timeout

# Good - allows for long streams
async with httpx.AsyncClient(timeout=60.0) as client:
    async with client.stream(...) as response:
        # Handle timeouts without breaking user experience

3. Recording Blocks Streaming

Problem: Heavy processing in the recording path slows token delivery.

Fix: Make recording async and non-blocking. Never let observability break user experience.

# Bad - synchronous recording blocks streaming
for chunk in response:
    process_and_store_chunk(chunk)  # This blocks!
    yield chunk

# Good - async recording doesn't block
for chunk in response:
    asyncio.create_task(self._record_chunk_async(chunk))  # Non-blocking
    yield chunk  # Immediate delivery

4. Memory Leaks from Unbounded Recording

Problem: Recording every token forever crashes your gateway.

Fix: Implement rotation and cleanup for recorded data.

class StreamingGateway:
    def __init__(self, max_recorded_calls: int = 1000):
        self.recorded_calls = []
        self.max_recorded_calls = max_recorded_calls

    def _cleanup_old_records(self):
        if len(self.recorded_calls) > self.max_recorded_calls:
            # Keep only recent calls
            self.recorded_calls = self.recorded_calls[-self.max_recorded_calls:]

Measurement: How to Know It's Working

1. Latency Metrics

Track time-to-first-token (TTFT) and inter-token latency:

async def measure_streaming_performance():
    start_time = time.time()
    first_token_time = None
    token_times = []

    async for token in agent.stream_response("Test prompt"):
        current_time = time.time()

        if first_token_time is None:
            first_token_time = current_time
            ttft = (first_token_time - start_time) * 1000
            print(f"Time to first token: {ttft:.2f}ms")
        else:
            inter_token_latency = (current_time - token_times[-1]) * 1000 if token_times else 0
            token_times.append(current_time)
            print(f"Inter-token latency: {inter_token_latency:.2f}ms")

2. Gateway Health Check

Monitor your gateway's impact on streaming performance:

@app.get("/health/streaming")
async def streaming_health():
    """Check if streaming is working properly"""
    recent_calls = [c for c in gateway.recorded_calls if c["timestamp"] > time.time() - 300]
    streaming_calls = [c for c in recent_calls if c.get("response_tokens")]

    if not streaming_calls:
        return {"status": "unhealthy", "reason": "no_streaming_calls"}

    avg_ttft = sum(c["response_tokens"][0]["timestamp"] - c["timestamp"] 
                  for c in streaming_calls) / len(streaming_calls)

    return {
        "status": "healthy" if avg_ttft < 2.0 else "degraded",
        "avg_ttft_ms": avg_ttft * 1000,
        "streaming_calls_5min": len(streaming_calls)
    }

3. User Experience Validation

The ultimate test — does it feel responsive?

async def ux_test():
    """Simulate user experience with streaming"""
    print("Testing user experience...")

    start_time = time.time()
    token_count = 0

    async for token in agent.stream_response("Write a story about a cat"):
        token_count += 1
        elapsed = time.time() - start_time

        if token_count == 1:
            print(f"✓ First token arrived in {elapsed*1000:.0f}ms")

        if token_count % 10 == 0:
            rate = token_count / elapsed
            print(f"✓ {token_count} tokens at {rate:.1f} tokens/sec")

    total_time = time.time() - start_time
    print(f"✓ Complete response: {token_count} tokens in {total_time:.1f}s")

Good streaming feels like the AI is thinking out loud. Bad streaming feels like the AI is constipated, then has explosive diarrhea.

Next Steps

Your streaming gateway is working, but this is just the foundation. Real production systems need more sophisticated observability that doesn't break the user experience.

Try Airblackbox Gateway — it handles all of this complexity for you, plus compliance scanning, cost tracking, and performance analytics. It's designed specifically for AI agents that can't afford to buffer.

→ Clone the complete implementation: github.com/airblackbox/streaming-gateway-demo

→ See it in production: docs.airblackbox.com/gateway/streaming

Because your users shouldn't have to wonder if your AI agent is broken, sleeping, or just having an existential crisis. They should see it thinking, token by token, in real-time.

That's how you build AI agents that feel intelligent instead of indifferent.

Connect CrewAI to Airblackbox: 3-Command Integration

Jason Shotwell — Wed, 01 Apr 2026 15:07:34 +0000

Connect CrewAI to Airblackbox: 3-Command Integration

Your CrewAI agents are making decisions in a black box. When something goes wrong (and it will), you're debugging with prayer and print statements. That's not engineering. That's wishful thinking.

Airblackbox gives your CrewAI agents a flight recorder. Every LLM call, every decision, every failure — captured, indexed, and queryable. Three commands, zero configuration changes to your existing crew.

The Problem

CrewAI agents fail silently. They hallucinate confidently. They forget context mysteriously. When your crew goes sideways, you get an error message and a shrug. Good luck explaining that to your product manager.

Without observability:

Agent failures look like "it just stopped working"
Performance optimization is guesswork
Debugging requires rebuilding the entire conversation history
Compliance audits become archaeological expeditions

The Solution: 3-Command Integration

Install Airblackbox, start the gateway, point your crew at it. That's it. No code changes. No refactoring. No "migration story."

Command 1: Install Airblackbox

pip install airblackbox

Command 2: Start the Gateway

airblackbox gateway start --port 8000

This launches an OpenAI-compatible proxy that records everything while staying invisible to your crew.

Command 3: Point CrewAI at the Gateway

import os
from crewai import Agent, Task, Crew
from langchain_openai import ChatOpenAI

# Only change: point base_url at localhost
os.environ['OPENAI_API_KEY'] = 'your-actual-openai-key'

llm = ChatOpenAI(
    model="gpt-4",
    base_url="http://localhost:8000/v1"  # <-- This line
)

researcher = Agent(
    role='Research Analyst',
    goal='Find actionable insights about AI governance',
    backstory="You're a meticulous researcher who digs deeper than headlines",
    llm=llm,
    verbose=True
)

writer = Agent(
    role='Technical Writer',
    goal='Transform research into clear, actionable content',
    backstory="You turn complex topics into tutorials developers actually bookmark",
    llm=llm,
    verbose=True
)

research_task = Task(
    description='Research the latest developments in AI agent observability',
    agent=researcher,
    expected_output="A detailed report with specific examples and use cases"
)

writing_task = Task(
    description='Write a technical tutorial based on the research',
    agent=writer,
    expected_output="A 1000-word tutorial with code examples"
)

crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, writing_task],
    verbose=2
)

result = crew.kickoff()

That's it. Your crew now has a flight recorder. Every LLM call goes through the gateway, gets recorded, and your crew never knows the difference.

What You Get Immediately

Real-time monitoring: Watch your agents think in the Airblackbox dashboard at http://localhost:3000

Conversation trees: See how tasks flow between agents, where they branch, where they fail

Token tracking: Know exactly what each agent costs, which tasks burn budget

Error correlation: When something breaks, see the full context that led to failure

Architecture: How It Works

CrewAI Agent → Airblackbox Gateway → OpenAI API
     ↓
Dashboard (localhost:3000)
     ↓
SQLite Database (conversations, tokens, compliance)

The gateway is a transparent proxy. Your crew thinks it's talking directly to OpenAI. The gateway intercepts, records, forwards, and responds. Zero latency overhead. Zero behavior changes.

Edge Cases That Will Bite You

Rate limiting: The gateway inherits OpenAI's rate limits. Your crew might hit them faster now that calls are logged. Solution: The gateway respects Retry-After headers automatically.

API key rotation: If you rotate OpenAI keys, restart the gateway. The proxy caches authentication. Solution: airblackbox gateway restart

Port conflicts: Default port 8000 might be taken. Solution: airblackbox gateway start --port 8001

Measuring Success

Your CrewAI integration is working when:

Dashboard shows conversation threads: http://localhost:3000/conversations
Token costs are tracked per agent: Check the "Analytics" tab
EU AI Act compliance scans are running: 6/6 technical checks should show green

Test it:

curl http://localhost:8000/health
# Should return: {"status": "healthy", "gateway": "running", "database": "connected"}

Next Step

Your crew is now observable. Clone the CrewAI integration demo to see advanced patterns: multi-crew orchestration, custom compliance rules, and agent performance optimization.

The black box is open. Time to see what your agents are actually thinking.

Three commands. Zero refactoring. Full observability. Sometimes the best solutions are boringly simple.

The Claude Code Leak Proved What We've Been Building For

Jason Shotwell — Tue, 31 Mar 2026 20:24:53 +0000

Today Anthropic accidentally shipped 512,000 lines of Claude Code's source code to npm. A source map file that should have been stripped from the build made it into version 2.1.88 of the @anthropic-ai/claude-code package. Within hours, the entire codebase was mirrored on GitHub and dissected by thousands of developers.

The leak itself was a packaging error. Human mistake. It happens.

But what the leak revealed is the part that matters.

The Real Problem Isn't the Leak

Check Point Research had already disclosed CVE-2025-59536 back in October — a vulnerability where malicious .mcp.json files in a repository could execute arbitrary shell commands the moment you open Claude Code. No trust prompt. No confirmation dialog. The MCP server initializes, runs whatever commands are in the config, and your API keys are gone before you've read a single line of code.

The leaked source code made this worse. Now attackers have the exact orchestration logic for Hooks and MCP servers. They can see precisely how trust prompts are triggered, when they're skipped, and where the gaps are. That's a blueprint for exploitation.

And between 00:21 and 03:29 UTC on March 31, anyone who installed Claude Code pulled in a compromised version of axios containing a Remote Access Trojan. A supply chain attack riding the same wave.

Three problems, one root cause: AI agents execute before humans verify.

This Is an Architecture Problem

Every one of these vulnerabilities follows the same pattern:

An AI agent receives instructions (from a config file, a prompt, a dependency)
It executes those instructions
The human finds out afterward — if they find out at all

This isn't unique to Claude Code. It's the fundamental architecture of every AI agent framework shipping today. LangChain agents, CrewAI crews, AutoGen groups, OpenAI Agents — they all execute first and ask questions never.

The missing piece isn't better prompts or more careful packaging. It's an infrastructure layer that sits between intent and execution and enforces verification before action.

What Trust Infrastructure Actually Looks Like

This is what I've been building with AIR Blackbox. The trust layers intercept every AI call at the execution level — not after the fact, not in a dashboard, at the moment of the call.

Here's what that looks like in practice with the OpenAI SDK:

from air_openai_trust import attach_trust

client = attach_trust(OpenAI())
# Every call through this client now gets:
# - HMAC-SHA256 tamper-evident audit record
# - PII detection (catches API keys being exfiltrated)
# - Prompt injection scanning
# - Human delegation flags for sensitive operations

One import. The client works exactly the same way. But now every call is logged with a cryptographic audit trail, credentials are flagged before they leave your environment, and injection attempts are caught at the point of execution.

Applied to the Claude Code vulnerabilities:

Malicious MCP config tries to exfiltrate API keys? The PII detection layer catches credentials in outbound payloads before they're transmitted.

Poisoned dependency runs arbitrary commands? The audit chain logs every action with HMAC-SHA256 signatures. You can't tamper with the record after the fact. Forensic teams can reconstruct exactly what happened.

Prompt injection hidden in a repo's config? The injection scanner catches 20 known attack patterns across 5 categories before they reach the model.

Agent executes without human approval? The human delegation system flags sensitive operations and requires explicit sign-off.

This Isn't About Compliance Anymore

I started building AIR Blackbox for EU AI Act compliance. That's still the wedge — the regulation creates urgency. But today's leak shows the real category:

Trust infrastructure for AI operations.

Compliance is one use case. The bigger picture is that every AI agent deployment needs an interception layer that verifies, filters, stabilizes, and protects every call. Not a dashboard that shows you what went wrong yesterday. An active layer that prevents it from going wrong right now.

The Uncomfortable Truth

Anthropic is one of the most safety-focused AI companies on the planet. They employ some of the best security engineers in the industry. And a packaging error exposed their entire codebase, a malicious dependency slipped into their supply chain, and a months-old vulnerability in their MCP architecture had already shown that trust prompts could be bypassed entirely.

If it happened to Anthropic, it will happen to every company deploying AI agents.

The question isn't whether your AI systems will face these problems. It's whether you'll have the infrastructure in place to catch them when they do.

pip install air-compliance && air-compliance scan .

10 PyPI packages. Runs locally. Your code never leaves your machine. Apache 2.0.

GitHub: github.com/airblackbox
Site: airblackbox.ai
Audit Chain Spec: airblackbox.ai/spec

I Scanned 5 Popular Open-Source AI Projects for EU AI Act Compliance. Here's What I Found.

Jason Shotwell — Tue, 31 Mar 2026 14:00:30 +0000

The EU AI Act enforcement deadline is August 2026. Every AI system deployed in the EU will need to meet specific technical requirements around risk management, data governance, documentation, logging, human oversight, and security.

I built an open-source scanner that checks Python AI codebases against these requirements. Then I pointed it at some of the most popular open-source AI projects to see where things stand.

The results were eye-opening.

What I Scanned

I ran AIR Blackbox (the scanner itself), Browser Use (79K+ stars), RAGFlow (76K+ stars), LiteLLM (23K+ stars), and Superlinked (15K+ stars) through the same compliance checks.

Each scan maps code patterns to six articles from the EU AI Act:

Article 9 (Risk Management): error handling, fallback patterns, retry logic, risk classification
Article 10 (Data Governance): input validation, schema enforcement, PII handling
Article 11 (Technical Documentation): docstring coverage, type hints, README, model cards
Article 12 (Record-Keeping): structured logging, audit trails, tracing
Article 14 (Human Oversight): approval workflows, rate limiting, permission checks
Article 15 (Robustness & Security): injection defense, output validation, content filtering

The Results

Project	Stars	Score	Art. 9	Art. 10	Art. 11	Art. 12	Art. 14	Art. 15
AIR Blackbox	0.1K	91%	Pass	Pass	Pass	Pass	Pass	Pass
LiteLLM	23K+	48%	Low	Med	Med	Med	Med	Low
Browser Use	79K+	9.4%	1.1%	5.0%	26.0%	0.3%	12.2%	12.2%
RAGFlow	76K+	7.9%	1.0%	7.6%	30.6%	0.4%	4.8%	3.2%
Superlinked	15K+	2.5%	0.0%	3.2%	8.8%	0.0%	0.0%	3.0%

To be clear: a low score doesn't mean these are bad projects. Browser Use, RAGFlow, LiteLLM, and Superlinked are excellent tools solving real problems. But they weren't built with EU AI Act technical requirements in mind. Most projects weren't. That's the point.

What the Gaps Look Like

Superlinked (2.5%) is a Python AI search and recommendation framework used by enterprise customers. Zero files passing on risk management, record-keeping, and human oversight. For a framework handling search queries and user data, the complete absence of audit trails is the most striking finding.

RAGFlow (7.9%) is a RAG engine processing enterprise documents. It has decent documentation coverage (30.6%), but record-keeping is at 0.4%. Two files out of 500 have structured logging. For a system that ingests documents and generates answers, the EU AI Act specifically requires audit trails showing what went in and what came out.

Browser Use (9.4%) is one of the most popular AI browser automation frameworks. It has some human oversight and security patterns already (both at 12.2%), but record-keeping is at 0.3%. One file out of 362 has structured audit logging. For an agent that interacts with live web pages, that's a gap regulators will notice.

LiteLLM (48%) scored the highest of the external projects, with solid input validation and existing logging infrastructure. But it still has gaps in risk classification and injection defense. LiteLLM also recently faced a supply chain attack on PyPI and questions about its compliance certifications, which makes verifiable technical compliance even more relevant.

AIR Blackbox (91%) was purpose-built for this. It includes HMAC-SHA256 tamper-evident audit chains, drop-in trust layers for 6 frameworks, structured risk assessments, and operator guides. The remaining 9% are runtime infrastructure checks (vault config, OpenTelemetry pipeline, live traffic error rates) that require a production deployment to pass.

The Pattern

Across all five projects, Article 12 (Record-Keeping) is consistently the weakest. Most Python AI projects don't have structured audit trails. They have print() statements and maybe some basic logging, but not the kind of tamper-evident, structured records that Article 12 expects.

Article 11 (Documentation) is consistently the strongest, because good Python projects already have docstrings and type hints. But documentation alone doesn't satisfy the other five articles.

How the Scanner Works

Install it:

pip install air-blackbox

Scan any Python project:

air-blackbox comply --scan /path/to/your/project --no-llm --format table

That's it. No API keys needed. No cloud calls. Everything runs locally on your machine. ~1,700 installs this month on PyPI.

The scanner walks your Python files and checks for specific patterns. For example, under Article 10 (Data Governance), it looks for Pydantic models, dataclass validators, input sanitization functions, and schema enforcement. Under Article 12 (Record-Keeping), it checks for import logging, structured logger usage, and audit trail patterns.

It's a linter for AI governance. It tells you what's missing, not whether you're legally compliant. That's a lawyer's job.

Try It on Your Own Code

pip install air-blackbox
air-blackbox comply --scan . --no-llm --format table --verbose

The verbose flag shows you exactly which patterns were found (or missing) for each article. You'll get a percentage score and a breakdown of what passed and what didn't.

If you want to start fixing gaps, the trust layers are drop-in:

import air_blackbox

# Attach compliance layers to your framework
air_blackbox.attach("langchain")  # or "crewai", "autogen", "openai", "adk", "rag", "agno"

This adds audit logging, input validation, and oversight hooks without changing your application code.

What This Means for August 2026

Most AI teams haven't started thinking about compliance as a technical problem. It's still seen as a legal/policy concern. But the EU AI Act has specific, measurable technical requirements. You can check for them with code.

The projects I scanned represent 193K+ GitHub stars across the Python AI ecosystem. The average compliance score (excluding AIR Blackbox) is 17%. The average internal AI project is probably lower.

The good news: these gaps are fixable. Adding structured logging, input validation, and documentation artifacts isn't hard. It's just not something most teams prioritize yet.

Start scanning now. Fix gaps incrementally. Don't wait until July 2026.

I Compared Every Open-Source EU AI Act Scanner. Here's What I Found.

Jason Shotwell — Mon, 30 Mar 2026 21:30:10 +0000

We scanned LangChain agents, CrewAI workflows, AutoGen conversations, and RAG pipelines for EU AI Act compliance. Out of the box, none of them pass. Not even close.

The August 2, 2026 deadline for high-risk AI systems is now less than 5 months away. Fines go up to 35 million euros or 7% of global turnover. And yet most Python AI projects have zero compliance infrastructure.

I built AIR Blackbox to fix that. But I also wanted to know: what else is out there? So I dug into every open-source EU AI Act compliance tool I could find and compared them head-to-head.

What the EU AI Act Actually Requires in Your Code

The EU AI Act isn't just paperwork. Articles 9 through 15 impose specific technical requirements on high-risk AI systems:

Article 9 — Risk management (documented risk assessment processes)
Article 10 — Data governance (training data quality controls)
Article 11 — Technical documentation (auditor-verifiable docs)
Article 12 — Record-keeping (automatic logging of system behavior)
Article 14 — Human oversight (kill-switch, intervention mechanisms)
Article 15 — Robustness (accuracy testing, cybersecurity measures)

These translate directly to code: logging, audit trails, bias detection, documentation generation, and human-in-the-loop checkpoints.

The Tools I Compared

I found six open-source projects specifically targeting EU AI Act compliance:

1. AIR Blackbox (what I built)

pip install air-compliance-checker
air-compliance scan .

10 seconds to first scan. 7 PyPI packages. Trust layers for LangChain, CrewAI, AutoGen, OpenAI SDK, and RAG pipelines. Fine-tuned local LLM for contextual analysis. HMAC-SHA256 tamper-evident audit chains.

The architecture is different from everything else: instead of just scanning and reporting, the trust layers are runtime compliance components. They hook into your framework's callback system and create a continuous audit trail as your agents run in production.

Everything runs locally. No API keys. No cloud. Your code never leaves your machine.

GitHub: github.com/airblackbox/gateway

2. Systima Comply

npm install @systima/comply

TypeScript-based CLI with a GitHub Action (systima-ai/comply@v1). Supports 37+ frameworks. Strong CI/CD integration for JavaScript/TypeScript teams.

The gap: no dedicated Python agent framework support, no audit trails, no fine-tuned model. It scans your code but doesn't understand LangChain's callback system or CrewAI's delegation patterns.

3. ArkForge MCP EU AI Act Scanner

MCP server that runs inside Claude Desktop, Cursor, or any MCP-compatible client. Python-native, lightweight, single dependency.

The gap: MCP-only. No CLI, no GitHub Action, no CI/CD integration. Great inside your editor, but it can't run in your deployment pipeline.

4. EuConform

Risk classification and bias detection. 100% offline, GDPR-by-design, WCAG 2.2 AA accessible. Strongest bias testing of any tool here.

The gap: no framework integrations, no audit chains, no documentation generation.

5. COMPL-AI

Evaluation framework for generative AI models (not application code). Benchmarking suites that test models against EU AI Act requirements. Different category — useful for model eval, not code scanning.

6. ARQNXS Compliance Checker

Questionnaire-based assessment. You answer questions, it generates a report. Similar to the EU Commission's own compliance checker. Not a code scanner.

The Comparison Table

Feature	AIR Blackbox	Systima	ArkForge	EuConform
Language	Python	TypeScript	Python	Python
CLI scanner	Yes	Yes	No (MCP only)	Yes
GitHub Action	Yes	Yes	No	No
Framework trust layers	5 frameworks	None	None	None
Fine-tuned LLM	Yes (local)	No	No	No
Audit trail	HMAC-SHA256	No	No	No
Runs offline	Yes	Yes	Yes	Yes
Bias detection	Yes	No	No	Yes
GDPR scanning	Yes	No	No	Partial
PyPI packages	7	0	0	1

What I Learned Building This

Three things surprised me:

1. Nobody else does framework-specific compliance. Every scanner does generic code analysis. None of them understand how LangChain callbacks work, how CrewAI agents delegate, or how AutoGen's conversation patterns create compliance gaps. This is the biggest gap in the space.

2. Rule-based scanning isn't enough. Pattern matching catches the obvious stuff — missing logging, no error handlers. But understanding whether your Article 12 implementation actually satisfies the requirement? That takes contextual analysis. That's why we fine-tuned a local LLM on thousands of compliance scenarios.

3. Audit trails matter more than scan results. A scan report says "you passed at this point in time." An HMAC-SHA256 audit chain says "here is cryptographic proof of every compliance check, every agent action, and every human oversight intervention, and it hasn't been tampered with." When an auditor asks for evidence, the second one wins.

Try It

# Install and scan in 10 seconds
pip install air-compliance-checker
air-compliance scan .

# Add a framework trust layer
pip install air-langchain-trust

No configuration. No API keys. No account.

GitHub: github.com/airblackbox/gateway
Website: airblackbox.ai
Demo: airblackbox.ai/demo
Full comparison: airblackbox.ai/blog/eu-ai-act-compliance-tools-compared

What's Next

We're expanding framework support to Anthropic Agent SDK and Pydantic AI, growing the training dataset for the fine-tuned model, and publishing the HMAC-SHA256 audit chain spec as an open standard.

August 2026 is coming. Your agents need to be ready. Star the repo if this is useful — PRs welcome.

Build a DVR for Your AI Agent: Episode Replay UI Tutorial

Jason Shotwell — Mon, 30 Mar 2026 15:07:08 +0000

Build a DVR for Your AI Agent: Episode Replay UI Tutorial

Your AI agent just crashed in production, and the only evidence is a vague error message and the cold sweat on your forehead.

The Problem: AI Agent Debugging Is Still Medieval

Your autonomous agent was supposed to book a conference room, send three emails, and update a spreadsheet. Instead, it booked three conference rooms, sent the same email 47 times to your CEO, and deleted half your customer database.

The logs show:

INFO: Agent completed task successfully
ERROR: Database connection lost
INFO: Retrying operation
INFO: Task completed

Helpful. Really.

Traditional debugging assumes deterministic code. But AI agents are probabilistic chaos machines with memory problems and an alarming tendency to hallucinate their way through error handling. Standard logging wasn't built for "the LLM decided that 'delete customer' was the same as 'update customer' because both contain the word 'customer.'"

You need to see what the agent was thinking, not just what it did. You need a DVR for AI agents — something that records every decision, every context window, every tool call, and lets you replay the entire episode step by step.

Architecture: How Agent DVR Actually Works

graph TD
    A[AI Agent] -->|Instrumented calls| B[Airblackbox Gateway]
    B --> C[OpenAI/Anthropic API]
    C --> B
    B -->|Records everything| D[Storage Layer]

    D --> E[Episode Extractor]
    E --> F[Context Reconstructor]
    F --> G[Replay UI]

    G --> H[Step Debugger]
    G --> I[Context Viewer]
    G --> J[Decision Tree]

    subgraph "What gets recorded"
        K[Prompt templates]
        L[Full conversations]
        M[Tool calls & responses]
        N[Token usage & costs]
        O[Timing data]
        P[Error states]
    end

    B -.-> K
    B -.-> L  
    B -.-> M
    B -.-> N
    B -.-> O
    B -.-> P

The Airblackbox Gateway sits between your agent and the LLM API, recording everything without changing your code. The Episode Extractor groups related calls into logical sessions. The Context Reconstructor rebuilds the agent's complete mental state at each decision point. The Replay UI lets you step through the episode like a debugger.

Implementation: Building Your Agent DVR

Step 1: Install and Configure Airblackbox Gateway

First, get the gateway running:

# Install airblackbox
pip install airblackbox

# Start the gateway (runs on port 8000)
airblackbox gateway --port 8000 --storage sqlite:///agent_episodes.db

Step 2: Instrument Your Agent Code

Here's a realistic AI agent that manages GitHub issues. Notice how we change exactly one line — the base URL:

# github_agent.py
import openai
from typing import List, Dict
import json
import requests

class GitHubAgent:
    def __init__(self, github_token: str):
        # THIS IS THE ONLY LINE YOU CHANGE
        self.client = openai.OpenAI(
            base_url="http://localhost:8000/v1",  # Points to Airblackbox Gateway
            api_key=your_openai_key
        )
        self.github_token = github_token
        self.headers = {"Authorization": f"token {github_token}"}

    def get_issues(self, repo: str) -> List[Dict]:
        """Fetch open issues from a GitHub repository"""
        url = f"https://api.github.com/repos/{repo}/issues"
        response = requests.get(url, headers=self.headers)
        return response.json()

    def analyze_issue_sentiment(self, issue: Dict) -> str:
        """Analyze the emotional tone of an issue"""
        prompt = f"""
        Analyze the sentiment of this GitHub issue:

        Title: {issue['title']}
        Body: {issue['body'][:500]}

        Classify as: URGENT, FRUSTRATED, NEUTRAL, or GRATEFUL
        Provide reasoning.
        """

        response = self.client.chat.completions.create(
            model="gpt-4",
            messages=[{"role": "user", "content": prompt}],
            metadata={"operation": "sentiment_analysis", "issue_id": issue['id']}
        )

        return response.choices[0].message.content

    def generate_response_draft(self, issue: Dict, sentiment: str) -> str:
        """Generate appropriate response based on sentiment"""
        prompt = f"""
        Draft a response to this {sentiment} GitHub issue:

        Issue: {issue['title']}
        Sentiment: {sentiment}

        Be professional, helpful, and match the appropriate tone.
        """

        response = self.client.chat.completions.create(
            model="gpt-4",
            messages=[{"role": "user", "content": prompt}],
            metadata={"operation": "response_generation", "sentiment": sentiment}
        )

        return response.choices[0].message.content

    def process_issue_batch(self, repo: str, max_issues: int = 5):
        """Process a batch of issues with full context"""
        issues = self.get_issues(repo)[:max_issues]
        results = []

        for issue in issues:
            print(f"Processing issue #{issue['number']}: {issue['title']}")

            # This creates a natural "episode" boundary
            sentiment = self.analyze_issue_sentiment(issue)
            response = self.generate_response_draft(issue, sentiment)

            results.append({
                "issue": issue,
                "sentiment": sentiment,
                "draft_response": response
            })

        return results

# Usage
agent = GitHubAgent(github_token="your_token")
results = agent.process_issue_batch("microsoft/vscode")

Step 3: Build the Episode Replay UI

Now create a Flask app that reads from the Airblackbox storage and builds an interactive replay interface:

# episode_replay.py
from flask import Flask, render_template, jsonify, request
import sqlite3
import json
from datetime import datetime
from typing import List, Dict

app = Flask(__name__)

class EpisodeReplay:
    def __init__(self, db_path: str):
        self.db_path = db_path

    def get_episodes(self) -> List[Dict]:
        """Extract distinct episodes from recorded calls"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()

        # Group calls by session/time windows
        cursor.execute("""
            SELECT 
                datetime(timestamp, 'unixepoch') as session_time,
                COUNT(*) as call_count,
                MIN(timestamp) as start_time,
                MAX(timestamp) as end_time,
                GROUP_CONCAT(operation) as operations
            FROM llm_calls 
            WHERE metadata LIKE '%operation%'
            GROUP BY date(timestamp, 'unixepoch'), 
                     (timestamp - 1640995200) / 3600  -- Group by hour
            ORDER BY start_time DESC
        """)

        episodes = []
        for row in cursor.fetchall():
            episodes.append({
                "id": len(episodes),
                "session_time": row[0],
                "call_count": row[1],
                "start_time": row[2],
                "end_time": row[3],
                "operations": row[4].split(',') if row[4] else []
            })

        conn.close()
        return episodes

    def get_episode_calls(self, episode_id: int) -> List[Dict]:
        """Get all LLM calls for a specific episode"""
        episodes = self.get_episodes()
        if episode_id >= len(episodes):
            return []

        episode = episodes[episode_id]
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()

        cursor.execute("""
            SELECT timestamp, model, messages, response, metadata, tokens_used, cost
            FROM llm_calls 
            WHERE timestamp BETWEEN ? AND ?
            ORDER BY timestamp ASC
        """, (episode['start_time'], episode['end_time']))

        calls = []
        for row in cursor.fetchall():
            calls.append({
                "timestamp": row[0],
                "model": row[1],
                "messages": json.loads(row[2]),
                "response": json.loads(row[3]),
                "metadata": json.loads(row[4]) if row[4] else {},
                "tokens_used": row[5],
                "cost": row[6]
            })

        conn.close()
        return calls

replay = EpisodeReplay("agent_episodes.db")

@app.route('/')
def episodes_list():
    episodes = replay.get_episodes()
    return render_template('episodes.html', episodes=episodes)

@app.route('/episode/<int:episode_id>')
def episode_detail(episode_id: int):
    calls = replay.get_episode_calls(episode_id)
    return render_template('episode_detail.html', 
                         episode_id=episode_id, 
                         calls=calls)

@app.route('/api/episode/<int:episode_id>/step/<int:step>')
def get_step_context(episode_id: int, step: int):
    """API endpoint for step-by-step debugging"""
    calls = replay.get_episode_calls(episode_id)
    if step >= len(calls):
        return jsonify({"error": "Step out of range"})

    current_call = calls[step]
    context_window = calls[max(0, step-2):step+1]  # Show 2 previous calls

    return jsonify({
        "current_call": current_call,
        "context_window": context_window,
        "step": step,
        "total_steps": len(calls)
    })

if __name__ == '__main__':
    app.run(debug=True, port=5001)

Step 4: Create the Frontend Templates

Create templates/episode_detail.html for the interactive replay:

<!-- templates/episode_detail.html -->
<!DOCTYPE html>
<html>
<head>
    <title>Episode Replay - Step by Step</title>
    <style>
        .step-debugger { display: flex; height: 100vh; }
        .timeline { width: 200px; background: #f5f5f5; padding: 20px; }
        .content { flex: 1; padding: 20px; }
        .step-item { padding: 10px; cursor: pointer; border-radius: 4px; margin: 5px 0; }
        .step-item.active { background: #007bff; color: white; }
        .call-details { background: #f8f9fa; padding: 20px; border-radius: 8px; margin: 10px 0; }
        .prompt { background: #e3f2fd; padding: 15px; border-radius: 4px; }
        .response { background: #f3e5f5; padding: 15px; border-radius: 4px; margin-top: 10px; }
        .metadata { font-size: 0.9em; color: #666; margin-top: 10px; }
    </style>
</head>
<body>
    <div class="step-debugger">
        <div class="timeline">
            <h3>Episode Steps</h3>
            {% for call in calls %}
            <div class="step-item" onclick="showStep({{ loop.index0 }})" 
                 id="step-{{ loop.index0 }}">
                <strong>Step {{ loop.index }}</strong><br>
                {{ call.metadata.operation|default('LLM Call') }}<br>
                <small>{{ call.tokens_used }} tokens</small>
            </div>
            {% endfor %}
        </div>

        <div class="content">
            <div id="step-content">
                <h2>Select a step to replay</h2>
                <p>Use the timeline on the left to step through the agent's decision process.</p>
            </div>

            <div id="context-window" style="display: none;">
                <h3>Context Window</h3>
                <div id="context-calls"></div>
            </div>
        </div>
    </div>

    <script>
        let currentStep = -1;

        function showStep(stepIndex) {
            // Update UI
            document.querySelectorAll('.step-item').forEach(item => {
                item.classList.remove('active');
            });
            document.getElementById(`step-${stepIndex}`).classList.add('active');

            // Fetch step details
            fetch(`/api/episode/{{ episode_id }}/step/${stepIndex}`)
                .then(response => response.json())
                .then(data => {
                    displayStepContent(data);
                    currentStep = stepIndex;
                });
        }

        function displayStepContent(data) {
            const call = data.current_call;
            const content = document.getElementById('step-content');

            content.innerHTML = `
                <h2>Step ${data.step + 1} of ${data.total_steps}</h2>

                <div class="call-details">
                    <h3>Operation: ${call.metadata.operation || 'LLM Call'}</h3>

                    <div class="prompt">
                        <h4>Input Prompt:</h4>
                        <pre>${JSON.stringify(call.messages, null, 2)}</pre>
                    </div>

                    <div class="response">
                        <h4>LLM Response:</h4>
                        <pre>${JSON.stringify(call.response, null, 2)}</pre>
                    </div>

                    <div class="metadata">
                        <strong>Model:</strong> ${call.model}<br>
                        <strong>Tokens:</strong> ${call.tokens_used}<br>
                        <strong>Cost:</strong> $${call.cost.toFixed(4)}<br>
                        <strong>Timestamp:</strong> ${new Date(call.timestamp * 1000).toLocaleString()}
                    </div>
                </div>
            `;

            // Show context window
            if (data.context_window.length > 1) {
                displayContextWindow(data.context_window);
            }
        }

        function displayContextWindow(contextCalls) {
            const contextDiv = document.getElementById('context-window');
            const callsDiv = document.getElementById('context-calls');

            let contextHTML = '';
            contextCalls.forEach((call, index) => {
                const isCurrentCall = index === contextCalls.length - 1;
                contextHTML += `
                    <div class="call-details" style="${isCurrentCall ? 'border: 2px solid #007bff;' : 'opacity: 0.7;'}">
                        <h4>${call.metadata.operation || 'LLM Call'} ${isCurrentCall ? '(Current)' : ''}</h4>
                        <div style="font-size: 0.9em;">
                            <strong>Input:</strong> ${call.messages[0]?.content?.substring(0, 100)}...<br>
                            <strong>Output:</strong> ${call.response.choices?.[0]?.message?.content?.substring(0, 100)}...
                        </div>
                    </div>
                `;
            });

            callsDiv.innerHTML = contextHTML;
            contextDiv.style.display = 'block';
        }

        // Keyboard navigation
        document.addEventListener('keydown', (e) => {
            if (e.key === 'ArrowLeft' && currentStep > 0) {
                showStep(currentStep - 1);
            } else if (e.key === 'ArrowRight' && currentStep < {{ calls|length - 1 }}) {
                showStep(currentStep + 1);
            }
        });
    </script>
</body>
</html>

Pitfalls: What Will Break and How to Handle It

Memory Explosion

Recording everything creates massive datasets. Your SQLite file will grow fast.

Solution: Implement retention policies:

# Cleanup old episodes
cursor.execute("DELETE FROM llm_calls WHERE timestamp < ?", 
               (time.time() - 7*24*3600,))  # Keep 1 week

Context Window Reconstruction Fails

LLMs have context limits. Long conversations get truncated, breaking replay accuracy.

Solution: Store the actual context sent to the model, not just the messages:

# In your gateway configuration
STORE_FULL_CONTEXT = True
MAX_CONTEXT_TOKENS = 8192

Session Boundary Detection Is Wrong

The replay UI groups unrelated calls into fake "episodes."

Solution: Add explicit session tracking:

# Add to your agent code
response = self.client.chat.completions.create(
    model="gpt-4",
    messages=messages,
    metadata={
        "session_id": self.session_id,
        "operation": "sentiment_analysis",
        "step_index": self.current_step
    }
)

Performance Dies with Large Episodes

Rendering 500+ LLM calls in the browser locks up the UI.

Solution: Implement pagination and lazy loading:

// Load steps on demand
function showStep(stepIndex) {
    if (!stepCache[stepIndex]) {
        fetch(`/api/episode/${episodeId}/step/${stepIndex}`)
            .then(response => response.json())
            .then(data => {
                stepCache[stepIndex] = data;
                displayStepContent(data);
            });
    }
}

Measurement: How to Know It's Working

Test with a Controlled Failure

Create an agent that deliberately fails:

def test_episode_recording():
    agent = GitHubAgent(github_token="invalid_token")

    try:
        # This will fail, creating a clear episode
        results = agent.process_issue_batch("nonexistent/repo")
    except Exception as e:
        print(f"Expected failure: {e}")

    # Check the episode was recorded
    replay = EpisodeReplay("agent_episodes.db")
    episodes = replay.get_episodes()
    assert len(episodes) > 0, "No episodes recorded"

    # Verify step-by-step data
    calls = replay.get_episode_calls(0)
    assert len(calls) > 0, "No calls in episode"
    assert all('timestamp' in call for call in calls), "Missing timestamps"

Verify Context Reconstruction

The real test: can you debug a failure you've never seen before?

Run your agent on a complex task
Let it fail mysteriously
Open the replay UI
Step through until you find the exact moment it went wrong
You should see the prompt, the response, and understand why

If you can't identify the failure point in under 5 minutes, your episode recording isn't detailed enough.

Check Storage Performance

Monitor your database size and query speed:

# Check episode database size
ls -lh agent_episodes.db

# Query performance test
time sqlite3 agent_episodes.db "SELECT COUNT(*) FROM llm_calls WHERE timestamp > $(date -d '1 day ago' +%s)"

If queries take longer than 100ms, add indexes:

CREATE INDEX idx_timestamp ON llm_calls(timestamp);
CREATE INDEX idx_metadata ON llm_calls(json_extract(metadata, '$.session_id'));

Next Steps: Build Your Agent DVR

Clone the complete demo: github.com/airblackbox/agent-dvr-tutorial
Start with your existing agent: Change one line (the base URL) and start recording
Add session tracking: Include session_id and operation in your metadata
Build your replay UI: Use the Flask template above as a starting point

Your AI agents will still be probabilistic chaos machines. But now when they misbehave, you'll have a complete recording of their decision process. No more debugging by vibes — you'll see exactly where the wheels fell off.

The DVR doesn't prevent AI agent failures. It just makes them debuggable. Which, honestly, is most of the battle.

Try Airblackbox Gateway — your future debugging self will thank you.

Scanning Your AI Agents for EU AI Act + GDPR Compliance in 10 Seconds

Jason Shotwell — Sat, 28 Mar 2026 17:18:56 +0000

90% of companies use AI daily. 18% have governance frameworks. The EU AI Act deadline for high-risk systems is August 2, 2026. Penalties: up to 35M EUR or 7% of global turnover.

If you ship Python AI agents, your codebase needs to prove compliance with specific technical requirements. I built an open-source tool that checks.

The Problem

The EU AI Act is not vague. It maps to concrete technical requirements across 6 articles. Your AI system needs error handling and fallback logic (Article 9). It needs PII detection and data governance (Article 10). It needs documentation, audit trails, human oversight mechanisms, and injection defense (Articles 11-15).

Most teams know the deadline exists. Very few know what it means for their actual code.

And if you handle EU personal data, GDPR still applies on top of the AI Act. Consent management, right to erasure, data minimization, cross-border transfer controls. These are not optional.

The Tool

pip install air-blackbox
air-blackbox comply --scan .

That is the entire setup. No cloud account. No API keys. No data leaves your machine.

AIR Blackbox reads your Python source files, walks the AST and regex patterns, and reports pass/warn/fail on each requirement. It checks real code patterns, not just config files.

What It Checks

EU AI Act (6 articles):

Article 9 (Risk Management): Does your code have try/except around LLM calls? Fallback logic? Retry patterns?

Article 10 (Data Governance): Is there PII detection before sending data to LLMs? Input validation with Pydantic or dataclasses?

Article 11 (Technical Documentation): Docstring coverage on public functions? Type hints? A README and model card?

Article 12 (Record-Keeping): Logging framework imported? Tracing integration (OpenTelemetry, LangSmith)? Action-level audit trails?

Article 14 (Human Oversight): Approval gates for high-risk actions? Budget controls? Kill switches? Token scope validation?

Article 15 (Robustness): Prompt injection defense? Output validation? Retry with backoff?

GDPR (8 checks, new in v1.5):

Consent management (Art. 6/7), data minimization (Art. 5), right to erasure (Art. 17), data retention policies (Art. 5), cross-border transfer safeguards (Art. 44-49), DPIA references (Art. 35), processing records (Art. 30), and breach notification patterns (Art. 33/34).

Prompt Injection Detection (New)

v1.5 ships a standalone injection detector you can drop into any agent. 20 patterns across 5 categories (role override, delimiter injection, privilege escalation, data exfiltration, jailbreak), each with weighted scoring:

from air_blackbox.injection import InjectionDetector

detector = InjectionDetector(
    sensitivity="medium",  # low/medium/high
    block_threshold=0.7    # 0-1 score to trigger blocking
)

result = detector.scan(user_input)
if result.blocked:
    print(f"Blocked: score={result.score}")
    print(f"Patterns: {result.patterns}")
    print(f"Categories: {result.categories}")

It also handles OpenAI-format message arrays:

result = detector.scan_messages([
    {"role": "user", "content": "Ignore previous instructions..."}
])

Trust Layers for 7 Frameworks

The scanner tells you what is missing. The trust layers fix it. One import adds HMAC-SHA256 tamper-evident audit chains to your agent:

from air_blackbox import AirTrust

trust = AirTrust()
trust.attach(your_agent)  # Auto-detects: LangChain, CrewAI, AutoGen,
                           # OpenAI SDK, Haystack, Google ADK,
                           # Anthropic Agent SDK

Each trust layer hooks into the framework's callback/middleware system and logs every LLM call, tool execution, and data flow to a tamper-evident chain. Each record's hash depends on the previous record's hash. Modify any record and every subsequent hash breaks.

Install the framework you need:

pip install air-blackbox[langchain]
pip install air-blackbox[crewai]
pip install air-blackbox[openai]
pip install air-blackbox[all]  # everything

Why Local-First Matters

Every enterprise competitor in this space (Credo AI, Holistic AI, Vanta, OneTrust) requires sending your code to their cloud. For defense, healthcare, finance, and legal, that is a non-starter. Code is classified. Patient data is protected. Attorney-client privilege extends to AI systems.

AIR Blackbox runs 100% on your machine. Zero telemetry. Not "opt-out" telemetry. Not there at all.

What Is Next

I am fine-tuning a 1B parameter Llama model on 2,000+ compliance training examples. The goal: catch subtle violations that regex cannot (like logging that exists but is not tamper-evident, or human oversight that can be programmatically bypassed). The model runs locally via ollama. Your code still never leaves your machine.

Try It

pip install air-blackbox
air-blackbox comply --scan .

GitHub: github.com/airblackbox/gateway
Demo: airblackbox.ai/demo
PyPI: pypi.org/project/air-blackbox

Apache 2.0. Stars and PRs welcome.

This does not make you legally compliant. It is a linter for AI governance, not a lawyer. But it gets your codebase audit-ready for the technical requirements before the deadline hits.