# I Upgraded My JSON Failure Library to SQLite. Now My AI Agents Share Crash Fixes Across Machines. > Source: > Published: 2026-07-24 16:22:27+00:00 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. SQLite is the perfect upgrade path. It's a single file, supports concurrent readers, and handles writes with proper locking. Here's the migration: ``` python 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: `idx_task_result` makes the lookup fast even with thousands of entries`timeout=30` prevents "database is locked" errors on WindowsSQLite 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. Moving from JSON to SQLite is a one-time migration. Here's how I did it without downtime: ``` python 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. Here's the updated MCPToolHandler with SQLite integration: ``` python 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. ``` 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: ``` python 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() ``` 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.