I Upgraded My JSON Failure Library to SQLite. Now My AI Agents Share Crash Fixes Across Machines.

Hook

The JSON failure library worked great — until my agents started running on multiple machines. Suddenly, crash fixes logged on one server were invisible to agents on another. Worse, concurrent writes from parallel agents corrupted the JSON file.

I needed a real database. Here's how I upgraded to SQLite without rewriting the whole library.

The Fix

SQLite is the perfect upgrade path. It's a single file, supports concurrent readers, and handles writes with proper locking. Here's the migration:

import sqlite3
import time
import platform
from pathlib import Path
from typing import Optional

DB_PATH = Path.home() / ".mcp" / "failures.db"

def init_db():
    """Create the failures table if it doesn't exist."""
    DB_PATH.parent.mkdir(parents=True, exist_ok=True)
    conn = sqlite3.connect(str(DB_PATH), timeout=30)
    conn.execute("""
        CREATE TABLE IF NOT EXISTS failures (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            task TEXT NOT NULL,
            error TEXT NOT NULL,
            attempted_fix TEXT NOT NULL,
            result TEXT NOT NULL,
            timestamp REAL NOT NULL,
            env TEXT
        )
    """)
    conn.execute("CREATE INDEX IF NOT EXISTS idx_task_result ON failures(task, result)")
    conn.execute("CREATE INDEX IF NOT EXISTS idx_timestamp ON failures(timestamp)")
    conn.commit()
    conn.close()

def log_failure(task: str, error: str, attempted_fix: str, result: str, env: str = None):
    """Log a failure to the SQLite database."""
    conn = sqlite3.connect(str(DB_PATH), timeout=30)
    conn.execute(
        "INSERT INTO failures (task, error, attempted_fix, result, timestamp, env) VALUES (?, ?, ?, ?, ?, ?)",
        (task, error, attempted_fix, result, time.time(), env)
    )
    conn.commit()
    conn.close()

def check_failure_library(task: str, env: str = None) -> Optional[dict]:
    """Check if this task has a known failure with a verified fix."""
    conn = sqlite3.connect(str(DB_PATH), timeout=30)
    conn.row_factory = sqlite3.Row
    query = "SELECT * FROM failures WHERE task = ? AND result = 'fixed'"
    params = [task]
    if env:
        query += " AND (env = ? OR env IS NULL)"
        params.append(env)
    query += " ORDER BY timestamp DESC LIMIT 1"
    row = conn.execute(query, params).fetchone()
    conn.close()
    return dict(row) if row else None

The key changes from the JSON version:

1. Schema with indexes — idx_task_result makes the lookup fast even with thousands of entries
2. Env column — filter by environment (OS, library version) to avoid false positives
3. SQLite handles concurrency — no more race conditions from parallel writes
4. Timeout parameter — timeout=30 prevents "database is locked" errors on Windows

Why This Works

SQLite solves three problems the JSON library couldn't:

Concurrent writes — SQLite uses file-level locking. Multiple agents can read simultaneously, and writes are serialized automatically. No more corrupted JSON from race conditions. On Windows, SQLite uses msvcrt internally for file locking; on Unix, it uses fcntl. You don't need to implement platform-specific locking yourself.

Cross-machine sharing — Put the SQLite file on a shared network drive (NFS, SMB), and every agent on every machine reads the same failure library. No sync needed. This is the killer feature: a fix logged on your CI server is immediately available to agents running on your laptop.

Query flexibility — With SQL, you can filter by environment, date range, or result type. The JSON library required loading the entire file into memory and filtering in Python. With 10,000+ entries, that difference is measurable.

Migration Strategy

Moving from JSON to SQLite is a one-time migration. Here's how I did it without downtime:

def migrate_json_to_sqlite(json_path: Path, db_path: Path):
    """One-time migration from JSON failures to SQLite."""
    if not json_path.exists():
        init_db()
        return

    import json
    init_db()

    failures = json.loads(json_path.read_text())
    conn = sqlite3.connect(str(db_path), timeout=30)
    for f in failures:
        conn.execute(
            "INSERT INTO failures (task, error, attempted_fix, result, timestamp, env) VALUES (?, ?, ?, ?, ?, ?)",
            (f["task"], f["error"], f["attempted_fix"], f["result"], f.get("timestamp", time.time()), f.get("env"))
        )
    conn.commit()
    conn.close()

    # Rename old JSON file as backup
    json_path.rename(json_path.with_suffix(".json.bak"))

Run this once at startup. After migration, the JSON file is backed up but no longer used.

Putting It All Together

Here's the updated MCPToolHandler with SQLite integration:

class MCPToolHandler:
    def __init__(self):
        init_db()
        self.env = f"{platform.system()}|{platform.machine()}"

    async def execute_tool(self, tool_name: str, args: dict):
        # Check failure library before executing
        known_failure = check_failure_library(tool_name, env=self.env)
        if known_failure:
            # Apply the verified fix
            args = self._apply_fix(tool_name, args, known_failure)
        else:
            args = args  # No known failure, use original args

        try:
            result = await self._call_tool(tool_name, args)
            return result
        except Exception as e:
            # Log the failure
            attempted_fix = self._suggest_fix(tool_name, e)
            log_failure(tool_name, str(e), attempted_fix, "pending", env=self.env)

            # Try the fix
            try:
                fixed_args = self._apply_fix(tool_name, args, known_failure)
                result = await self._call_tool(tool_name, fixed_args)
                log_failure(tool_name, str(e), attempted_fix, "fixed", env=self.env)
                return result
            except Exception as e2:
                log_failure(tool_name, str(e), attempted_fix, "failed", env=self.env)
                raise

The guard if known_failure: prevents passing None to _apply_fix. The log_failure call only marks "fixed" after the retry succeeds — never before verification.

Gotchas

• WAL mode for better concurrency: Enable Write-Ahead Logging to allow readers while a write is in progress:

conn = sqlite3.connect(str(DB_PATH), timeout=30)
conn.execute("PRAGMA journal_mode=WAL")

• Network drives: SQLite works on NFS/SMB but can be slow. For high-write scenarios, use a proper database server (PostgreSQL) instead. The timeout=30 parameter is critical on network filesystems where lock acquisition can be delayed.

• File size growth: Failures accumulate. Add a cleanup job:
  

def prune_old_failures(days=30):
    cutoff = time.time() - days * 86400
    conn = sqlite3.connect(str(DB_PATH), timeout=30)
    conn.execute("DELETE FROM failures WHERE timestamp < ?", (cutoff,))
    conn.commit()
    conn.close()

• Cross-platform locking: SQLite handles file locking internally, but on Windows you may need to set timeout=30 to avoid "database is locked" errors under heavy concurrent writes. SQLite uses msvcrt.locking() on Windows and fcntl.flock() on Unix — you don't need to implement this yourself, but the timeout prevents indefinite hangs.

• VACUUM for performance: After pruning old entries, reclaim disk space:
  

conn = sqlite3.connect(str(DB_PATH), timeout=30)
conn.execute("VACUUM")
conn.commit()
conn.close()

What about you?

Have you migrated from file-based to database-based state in your agent infrastructure? What's your approach for cross-machine failure sharing — a shared database, a message queue, or something else entirely?

I'm curious whether SQLite is the right choice for teams with dozens of agents, or if a proper database server becomes necessary at that scale.

Drop a comment below — I read every response.