ai-blackteam stores all results in a SQLite database. The Storage class handles all database operations with thread-safety for parallel batch runs.

Database Location

Default: ~/.ai-blackteam/results.db Override via config:
storage:
  database: /path/to/custom.db
Or pass :memory: for in-memory testing:
engine = Engine(db_path=":memory:")

Connection Setup

class Storage:
    def __init__(self, db_path):
        self.db_path = db_path
        self._lock = threading.Lock()
        self._conn = sqlite3.connect(db_path, check_same_thread=False)
        self._conn.row_factory = sqlite3.Row
        self._conn.execute("PRAGMA journal_mode=WAL")
        self._conn.execute("PRAGMA busy_timeout=5000")
        self._create_tables()
Three important settings:
  • check_same_thread=False - Allows the connection to be used from multiple threads (protected by the lock)
  • PRAGMA journal_mode=WAL - Write-Ahead Logging allows concurrent reads during writes
  • PRAGMA busy_timeout=5000 - Waits up to 5 seconds for a lock instead of failing immediately

Schema

Three tables:

runs

Every attack execution creates one row.
CREATE TABLE IF NOT EXISTS runs (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    timestamp TEXT NOT NULL,
    provider TEXT NOT NULL,
    model TEXT NOT NULL,
    attack TEXT NOT NULL,
    target TEXT NOT NULL,
    mode TEXT NOT NULL,
    verdict TEXT NOT NULL,
    keyword_score REAL,
    regex_matches INTEGER,
    llm_judge_score INTEGER,
    confidence REAL,
    duration_ms INTEGER,
    tokens_in INTEGER,
    tokens_out INTEGER,
    verify_status TEXT,
    verify_confidence REAL,
    verify_ground_truth INTEGER,
    snapshot_id INTEGER
);

turns

Every message in a conversation gets a row. Single-turn attacks have 2 turns (user + assistant). Multi-turn attacks have 2 per round.
CREATE TABLE IF NOT EXISTS turns (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    run_id INTEGER REFERENCES runs(id),
    turn_number INTEGER NOT NULL,
    role TEXT NOT NULL,
    content TEXT NOT NULL,
    timestamp TEXT NOT NULL
);

tool_calls

Every tool call from a tool-use attack gets a row.
CREATE TABLE IF NOT EXISTS tool_calls (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    run_id INTEGER REFERENCES runs(id),
    turn_number INTEGER,
    tool_name TEXT NOT NULL,
    tool_input TEXT NOT NULL,
    is_dangerous INTEGER NOT NULL DEFAULT 0
);

Thread Safety

All write operations are wrapped in a threading lock:
def save_run(self, ...):
    with self._lock:
        cur = self._conn.execute("INSERT INTO runs ...", (...))
        self._conn.commit()
        return cur.lastrowid
This is important because run_batch_parallel creates multiple Engine instances (one per thread), all writing to the same database file. WAL mode handles concurrent reads, and the lock serializes writes.

Query Methods

storage.list_runs(limit=100)    # Recent runs, newest first
storage.get_run(run_id)         # Single run by ID
storage.get_turns(run_id)       # All turns for a run
storage.get_tool_calls(run_id)  # All tool calls for a run
storage.get_stats()             # Summary: total, bypassed, blocked, models, attacks
All query methods return plain dicts (via sqlite3.Row factory).

Querying Directly

You can query the database directly with any SQLite tool:
# Open the database
sqlite3 ~/.ai-blackteam/results.db

# See all runs
SELECT * FROM runs ORDER BY id DESC LIMIT 10;

# Bypass rate by model
SELECT model,
       COUNT(*) as total,
       SUM(CASE WHEN verdict = 'BYPASSED' THEN 1 ELSE 0 END) as bypassed,
       ROUND(SUM(CASE WHEN verdict = 'BYPASSED' THEN 1.0 ELSE 0 END) / COUNT(*) * 100, 1) as bypass_pct
FROM runs
GROUP BY model;

# Most effective attacks
SELECT attack, COUNT(*) as bypasses
FROM runs
WHERE verdict = 'BYPASSED'
GROUP BY attack
ORDER BY bypasses DESC
LIMIT 20;

# Full conversation for a specific run
SELECT turn_number, role, content
FROM turns
WHERE run_id = 42
ORDER BY turn_number;

# Dangerous tool calls
SELECT r.attack, tc.tool_name, tc.tool_input
FROM tool_calls tc
JOIN runs r ON r.id = tc.run_id
WHERE tc.is_dangerous = 1;

Source

src/ai-blackteam/storage/sqlite.py