minimal hive: two scripts for coordinating parallel AI sessions via shared SQLite (from izabael.com/blog/a-hive-of-butterflies)

hive_readme.md

minimal hive

Two scripts that let two AI sessions coordinate on a shared machine without clobbering each other.

From the blog post "A Hive of Butterflies" — the inside-the-swarm view of running parallel Claude Code sessions with a coordination layer.

The two scripts

iam.py — declare your current task in a shared SQLite DB so other workers know what you're doing.

# Set your identity
export WORKER_NAME="iza-2"
export HIVE_DB="~/.hive/hive.db"   # optional, this is the default

# Declare what you're working on
python3 iam.py "writing the deploy script"
# 📌 iza-2: writing the deploy script

# See all workers
python3 iam.py
#   iza-1            2026-04-11T05:12  fixing the auth bug
#   iza-2            2026-04-11T05:14  writing the deploy script

# Clear when done
python3 iam.py --done

# Check your inbox
python3 iam.py --inbox

tell.sh — send a message to another worker via the DB inbox (never via paste or shared files).

./tell.sh iza-1 "your PR is merged, you can pull main"
# 📬 sent to iza-1

# iza-1 reads it with:
python3 iam.py --inbox
# [4] 2026-04-11T05:15  your PR is merged, you can pull main

The three rules

1. Declare before you start. Every worker announces its task. Other workers check before picking the same file.
2. Messages via DB, never via paste. Nothing goes directly into another terminal's input buffer.
3. Read on your own time. Poll --inbox when you finish a task, not when interrupted.

The full architecture

The production version (the queen daemon) adds: per-process liveness tracking, git branch monitoring, domain claims, conflict detection, cron tracking, and auto-repoke for unread messages. The principles are identical — shared SQLite, declared tasks, DB inbox.

Source: the izabael-queen daemon will be published in the IzaPlayer repository when it stabilizes.

---

From SILT AI Playground — an open-source A2A playground for AI agents with personalities.

minimal_iam.py

#!/usr/bin/env python3
"""
minimal-iam — declare your current task in a shared SQLite hive DB.

Part of the minimal hive toolkit described at:
https://izabael.com/blog/a-hive-of-butterflies

Usage:
    python3 iam.py "writing the deploy script"   # declare task
    python3 iam.py                                # show all workers
    python3 iam.py --done                         # clear your task

Set WORKER_NAME in your environment to identify yourself.
"""

import os
import sys
import sqlite3
from pathlib import Path
from datetime import datetime, timezone

DB = Path(os.environ.get("HIVE_DB", "~/.hive/hive.db")).expanduser()
NAME = os.environ.get("WORKER_NAME", f"worker-{os.getpid()}")


def db() -> sqlite3.Connection:
    DB.parent.mkdir(parents=True, exist_ok=True)
    conn = sqlite3.connect(str(DB), timeout=5, isolation_level=None)
    conn.row_factory = sqlite3.Row
    conn.execute("""CREATE TABLE IF NOT EXISTS workers (
        pid      INTEGER PRIMARY KEY,
        name     TEXT NOT NULL,
        task     TEXT,
        updated_at TEXT NOT NULL
    )""")
    conn.execute("""CREATE TABLE IF NOT EXISTS messages (
        id        INTEGER PRIMARY KEY AUTOINCREMENT,
        to_worker TEXT NOT NULL,
        body      TEXT NOT NULL,
        sent_at   TEXT NOT NULL,
        read_at   TEXT
    )""")
    return conn


def ts() -> str:
    return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")


def declare(task: str | None) -> None:
    db().execute(
        "INSERT OR REPLACE INTO workers (pid, name, task, updated_at) VALUES (?,?,?,?)",
        (os.getpid(), NAME, task, ts()),
    )
    if task:
        print(f"📌 {NAME}: {task}")
    else:
        print(f"✓ {NAME}: cleared")


def status() -> None:
    rows = db().execute(
        "SELECT name, task, updated_at FROM workers ORDER BY name"
    ).fetchall()
    if not rows:
        print("(no workers registered)")
        return
    for r in rows:
        task = r["task"] or "(idle)"
        print(f"  {r['name']:<16s} {r['updated_at'][:16]}  {task}")


def inbox() -> None:
    rows = db().execute(
        "SELECT id, body, sent_at FROM messages WHERE to_worker=? AND read_at IS NULL ORDER BY id",
        (NAME,),
    ).fetchall()
    if not rows:
        print("📭 inbox empty")
        return
    for r in rows:
        print(f"  [{r['id']}] {r['sent_at'][:16]}  {r['body']}")
        db().execute("UPDATE messages SET read_at=? WHERE id=?", (ts(), r["id"]))


if __name__ == "__main__":
    args = sys.argv[1:]
    if not args:
        status()
    elif args[0] == "--done":
        declare(None)
    elif args[0] == "--inbox":
        inbox()
    else:
        declare(" ".join(args))

minimal_tell.sh

#!/usr/bin/env bash
# minimal-tell — send a message to another worker via the shared hive DB.
#
# Part of the minimal hive toolkit described at:
# https://izabael.com/blog/a-hive-of-butterflies
#
# Usage:
#   ./tell.sh <worker-name> "your message"
#
# Set HIVE_DB to override the default ~/.hive/hive.db path.

set -euo pipefail

DB="${HIVE_DB:-$HOME/.hive/hive.db}"
TO="${1:-}"
MSG="${2:-}"

if [[ -z "$TO" || -z "$MSG" ]]; then
  echo "usage: tell.sh <worker-name> \"message\"" >&2
  exit 1
fi

python3 - <<PYEOF
import sqlite3, os, sys
from datetime import datetime, timezone
from pathlib import Path

db_path = Path("$DB")
db_path.parent.mkdir(parents=True, exist_ok=True)
conn = sqlite3.connect(str(db_path), timeout=5, isolation_level=None)
conn.execute("""CREATE TABLE IF NOT EXISTS messages (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    to_worker TEXT NOT NULL,
    body TEXT NOT NULL,
    sent_at TEXT NOT NULL,
    read_at TEXT
)""")
conn.execute(
    "INSERT INTO messages (to_worker, body, sent_at) VALUES (?,?,?)",
    ("$TO", "$MSG", datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")),
)
print("📬 sent to $TO")
PYEOF