Anthropic's always-on agents guide: what the inbox, timers and outbox do when a worker dies, tested in SQLite

"""Inbox, lease, timers, approvals and outbox for an always-on agent.
 
One SQLite file per user: the file is the user's inbox. Times are integer
seconds, and every function takes `now` so the rules can be tested without
sleeping.
"""
import json
import sqlite3
from contextlib import contextmanager
 
MAX_ATTEMPTS = 3      # claims before an event is parked as dead
LEASE_TTL = 120       # seconds a worker owns the inbox without renewing
OUTBOX_GAP = 3600     # at most one message per hour
 
SCHEMA = """
CREATE TABLE IF NOT EXISTS inbox (
  id INTEGER PRIMARY KEY,
  source TEXT NOT NULL,
  source_id TEXT NOT NULL,
  payload TEXT,
  status TEXT NOT NULL DEFAULT 'pending',   -- pending | running | done | dead
  attempts INTEGER NOT NULL DEFAULT 0,
  claim_gen INTEGER,                        -- lease generation that claimed it
  error TEXT,
  UNIQUE (source, source_id)
);
CREATE TABLE IF NOT EXISTS lease (
  one INTEGER PRIMARY KEY CHECK (one = 1),  -- a single row
  holder TEXT NOT NULL,
  gen INTEGER NOT NULL,                     -- +1 on every takeover
  until INTEGER NOT NULL
);
CREATE TABLE IF NOT EXISTS timers (
  id TEXT PRIMARY KEY,
  due INTEGER NOT NULL,
  every INTEGER,                            -- NULL = one-off follow-up
  payload TEXT,
  ev_source TEXT, ev_id TEXT,               -- fixed event key (one-offs only)
  done INTEGER NOT NULL DEFAULT 0
);
CREATE TABLE IF NOT EXISTS outbox (
  id INTEGER PRIMARY KEY,
  body TEXT NOT NULL,
  created INTEGER NOT NULL,
  sent_at INTEGER
);
"""
 
 
def connect(path):
    db = sqlite3.connect(path, isolation_level=None)  # transactions are explicit
    db.execute("PRAGMA journal_mode=WAL")
    db.executescript(SCHEMA)
    return db
 
 
@contextmanager
def tx(db):
    """A write transaction that always ends: COMMIT on success, ROLLBACK on error."""
    db.execute("BEGIN IMMEDIATE")
    try:
        yield
    except BaseException:
        db.execute("ROLLBACK")
        raise
    db.execute("COMMIT")
 
 
# --- inbox -----------------------------------------------------------------
 
def ingest(db, source, source_id, payload):
    """Store an event. Call it after the webhook signature check, then ack.
 
    Returns False for a duplicate delivery: the event is stored once.
    """
    cur = db.execute(
        "INSERT OR IGNORE INTO inbox (source, source_id, payload) VALUES (?, ?, ?)",
        (source, source_id, json.dumps(payload)))
    return cur.rowcount == 1
 
 
# --- one worker per inbox --------------------------------------------------
 
def acquire(db, holder, now):
    """Take or renew the lease. Returns the generation, or None if someone else holds it.
 
    Renewing keeps the generation; any takeover, even by the same holder
    after its lease lapsed, starts a new one.
    """
    with tx(db):
        row = db.execute("SELECT holder, gen, until FROM lease").fetchone()
        if row is None:
            gen = 1
        elif row[0] == holder and row[2] >= now:
            gen = row[1]                         # renewal
        elif row[2] < now:
            gen = row[1] + 1                     # takeover of an expired lease
        else:
            return None                          # held by someone else
        db.execute("INSERT OR REPLACE INTO lease VALUES (1, ?, ?, ?)",
                   (holder, gen, now + LEASE_TTL))
        return gen
 
 
def _holds(db, holder, gen, now):
    row = db.execute("SELECT holder, gen, until FROM lease").fetchone()
    return row is not None and row[:2] == (holder, gen) and row[2] >= now
 
 
def claim_batch(db, holder, gen, now):
    """Claim work for one run. Call once per run, after the previous run's finish().
 
    Rows left 'running' by an older generation belong to a worker that lost
    its lease: they go back to pending. Attempts are counted at claim time, so
    an event that crashes the worker every time still runs out of attempts.
    A retried event is claimed alone, so a poison event can't take healthy
    events down with it.
    """
    with tx(db):
        if not _holds(db, holder, gen, now):
            return []
        db.execute("""UPDATE inbox SET status = 'pending'
                      WHERE status = 'running' AND claim_gen < ?""", (gen,))
        db.execute("""UPDATE inbox SET status = 'dead', error = 'too many attempts'
                      WHERE status = 'pending' AND attempts >= ?""", (MAX_ATTEMPTS,))
        retry = db.execute("""SELECT id FROM inbox WHERE status = 'pending'
                              AND attempts > 0 ORDER BY id LIMIT 1""").fetchone()
        where = "id = ?" if retry else "status = 'pending'"
        rows = db.execute(
            f"""UPDATE inbox SET status = 'running', attempts = attempts + 1,
                    claim_gen = ? WHERE {where}
                RETURNING id, source, source_id, payload""",
            (gen, *retry) if retry else (gen,)).fetchall()
    return sorted(rows)                          # RETURNING order is arbitrary
 
 
def finish(db, holder, gen, now, ids, messages=(), error=None):
    """Record a run's outcome and queue its messages in one fenced transaction.
 
    Refused (False) unless this holder still holds the lease in the same
    generation that claimed the events, and the events are still running.
    """
    with tx(db):
        if not _holds(db, holder, gen, now):
            return False
        status = "pending" if error else "done"
        for i in ids:
            db.execute("""UPDATE inbox SET status = ?, error = ?
                          WHERE id = ? AND status = 'running' AND claim_gen = ?""",
                       (status, error, i, gen))
        if not error:
            for body in messages:
                db.execute("INSERT INTO outbox (body, created) VALUES (?, ?)",
                           (body, now))
        return True
 
 
# --- timers ----------------------------------------------------------------
 
def add_timer(db, timer_id, due, payload, every=None, key=(None, None)):
    if every is not None and every <= 0:
        raise ValueError("every must be a positive number of seconds")
    if every is not None and key != (None, None):
        raise ValueError("a recurring timer can't use a fixed event key")
    db.execute(
        """INSERT OR REPLACE INTO timers (id, due, every, payload, ev_source, ev_id)
           VALUES (?, ?, ?, ?, ?, ?)""",
        (timer_id, int(due), every, json.dumps(dict(payload)), *key))
 
 
def run_scheduler(db, now):
    """Turn due timers into inbox events. Never starts a run itself.
 
    The timer advances in the same transaction as the insert, so a second pass
    finds nothing due. The event key (timer id + scheduled time) is the second
    line of defence: a replay of the same occurrence is a duplicate. Missed
    occurrences coalesce into one event for the latest of them.
    """
    with tx(db):
        due = db.execute(
            """SELECT id, due, every, payload, ev_source, ev_id FROM timers
               WHERE done = 0 AND due <= ?""", (now,)).fetchall()
        for timer_id, at, every, payload, ev_source, ev_id in due:
            skipped = (now - at) // every if every else 0
            at += skipped * (every or 0)
            body = json.loads(payload)
            body.update(scheduled_for=at, late_by=now - at, skipped=skipped)
            db.execute(
                "INSERT OR IGNORE INTO inbox (source, source_id, payload) VALUES (?, ?, ?)",
                (ev_source or "timer", ev_id or f"{timer_id}@{at}", json.dumps(body)))
            if every:
                db.execute("UPDATE timers SET due = ? WHERE id = ?", (at + every, timer_id))
            else:
                db.execute("UPDATE timers SET done = 1 WHERE id = ?", (timer_id,))
    return len(due)
 
 
# --- deferred approvals ----------------------------------------------------
 
def request_approval(db, request_id, now, expires_in, action):
    """The run ends here. The answer, or its expiry, comes back as an inbox event
    under the same key ('approval', request_id): only one of them is ever stored.
    The timer resumes the session when nobody answers."""
    add_timer(db, f"expire:{request_id}", now + expires_in,
              {"approval": request_id, "answer": "expired", "action": action},
              key=("approval", request_id))
 
 
def answer_approval(db, request_id, answer, now):
    """Store the user's answer unless the deadline passed. Returns the outcome
    that stands: 'yes'/'no', 'expired', or 'unknown' for a request never made."""
    with tx(db):
        row = db.execute("SELECT due, payload FROM timers WHERE id = ?",
                         (f"expire:{request_id}",)).fetchone()
        if row is None:
            return "unknown"
        due, expiry = row
        payload = expiry if now >= due else json.dumps(
            {"approval": request_id, "answer": answer})
        db.execute("""INSERT OR IGNORE INTO inbox (source, source_id, payload)
                      VALUES ('approval', ?, ?)""", (request_id, payload))
        stored = db.execute("""SELECT payload FROM inbox
                               WHERE source = 'approval' AND source_id = ?""",
                            (request_id,)).fetchone()[0]
    return json.loads(stored)["answer"]
 
 
# --- outbox ----------------------------------------------------------------
 
def dispatch(db, now, send):
    """Send the held messages as one, at most once per OUTBOX_GAP.
 
    Runs inside the write transaction, so two dispatchers can't both send the
    same rows. A crash after send() and before COMMIT sends them again on the
    next pass: delivery is at-least-once, so the channel should dedupe or the
    duplicate has to be harmless.
    """
    with tx(db):
        last = db.execute("SELECT MAX(sent_at) FROM outbox").fetchone()[0]
        if last is not None and now - last < OUTBOX_GAP:
            return 0
        held = db.execute(
            "SELECT id, body FROM outbox WHERE sent_at IS NULL ORDER BY id").fetchall()
        if not held:
            return 0
        send("\n\n".join(body for _, body in held))
        db.executemany("UPDATE outbox SET sent_at = ? WHERE id = ?",
                       [(now, i) for i, _ in held])
    return len(held)

"""Failure scenarios for inbox.py. Simulated clock, one process; a 'crash' is a
worker that claims and never calls finish()."""
import json
import os
import tempfile
 
import inbox as ib
 
db = ib.connect(os.path.join(tempfile.mkdtemp(), "agent.db"))
t = 1_000_000
DAY = 86400
 
 
def run(holder, now, messages=()):
    """One run: take the lease, claim, finish."""
    gen = ib.acquire(db, holder, now)
    batch = ib.claim_batch(db, holder, gen, now)
    ib.finish(db, holder, gen, now, [r[0] for r in batch], messages)
    return batch
 
 
def status(source_id):
    return db.execute("SELECT status, attempts FROM inbox WHERE source_id = ?",
                      (source_id,)).fetchone()
 
 
print("1. webhook delivered twice")
print("   first:", ib.ingest(db, "github", "delivery-7f3a", {"pr": 12}),
      " retry:", ib.ingest(db, "github", "delivery-7f3a", {"pr": 12}),
      " rows:", db.execute("SELECT COUNT(*) FROM inbox").fetchone()[0])
 
print("2. two workers ask for the lease at the same time")
print("   A:", ib.acquire(db, "A", t), " B:", ib.acquire(db, "B", t))
run("A", t)
 
print("3. three quick chat messages")
for n in range(3):
    ib.ingest(db, "chat", f"msg-{n}", {"text": f"message {n}"})
print("   events in one run:", len(run("A", t + 5)))
 
print("4. A stalls mid-run; B takes over; A wakes up")
ib.ingest(db, "chat", "msg-3", {"text": "send the report"})
gen_a = ib.acquire(db, "A", t + 10)
batch_a = ib.claim_batch(db, "A", gen_a, t + 10)          # A stalls here
print("   B before A's lease expires:", ib.acquire(db, "B", t + 60))
t2 = t + 10 + ib.LEASE_TTL + 1
gen_b = ib.acquire(db, "B", t2)
batch_b = ib.claim_batch(db, "B", gen_b, t2)
print("   B after expiry: gen", gen_b, " re-claims", [r[2] for r in batch_b])
ib.finish(db, "B", gen_b, t2, [r[0] for r in batch_b], ["report sent"])
t3 = t2 + ib.LEASE_TTL + 1                                 # B's lease lapses too
print("   A renews after both leases lapsed: gen", ib.acquire(db, "A", t3))
print("   A finishes its old claim (gen", gen_a, "):",
      ib.finish(db, "A", gen_a, t3, [r[0] for r in batch_a], ["report sent"]))
print("   msg-3:", status("msg-3"), " outbox rows:",
      db.execute("SELECT COUNT(*) FROM outbox").fetchone()[0])
 
print("5. a poison event arrives together with a healthy one")
ib.ingest(db, "chat", "poison", {"text": "x"})
ib.ingest(db, "chat", "healthy", {"text": "ok"})
now = t3 + ib.LEASE_TTL + 1
for n in range(5):
    gen = ib.acquire(db, f"W{n}", now)
    got = ib.claim_batch(db, f"W{n}", gen, now)
    ids = [r[2] for r in got]
    if "poison" in ids:
        print(f"   W{n} claimed {ids} and crashed")   # no finish()
    else:
        ib.finish(db, f"W{n}", gen, now, [r[0] for r in got])
        print(f"   W{n} claimed {ids} and finished")
    now += ib.LEASE_TTL + 1
print("   poison:", status("poison"), " healthy:", status("healthy"))
 
print("6. timers: a second scheduler pass, a replay, a three-day outage")
ib.add_timer(db, "digest", now, {"task": "daily digest"}, every=DAY)
print("   pass 1:", ib.run_scheduler(db, now), " pass 2:", ib.run_scheduler(db, now))
ib.add_timer(db, "digest", now, {"task": "daily digest"}, every=DAY)  # replay
print("   replay of the same occurrence:", ib.run_scheduler(db, now), "due,",
      db.execute("SELECT COUNT(*) FROM inbox WHERE source_id LIKE 'digest@%'")
      .fetchone()[0], "event total")
later = now + DAY + 3 * DAY + 600                         # next due + 3 days
ib.run_scheduler(db, later)
ev = json.loads(db.execute("""SELECT payload FROM inbox WHERE source_id LIKE 'digest@%'
                              ORDER BY id DESC LIMIT 1""").fetchone()[0])
print("   after the outage: events total",
      db.execute("SELECT COUNT(*) FROM inbox WHERE source_id LIKE 'digest@%'")
      .fetchone()[0], " skipped", ev["skipped"], " late by", ev["late_by"] // 60, "min",
      " next due in", round((db.execute("SELECT due FROM timers WHERE id = 'digest'")
                             .fetchone()[0] - later) / 3600, 1), "h")
 
print("7. approvals with a four-hour window")
h = 3600
ib.request_approval(db, "refund-881", later, 4 * h, "refund order 881")
ib.run_scheduler(db, later + 5 * h)
print("   881: expiry ran at 5h, yes at 6h ->", ib.answer_approval(db, "refund-881", "yes", later + 6 * h))
ib.request_approval(db, "refund-882", later, 4 * h, "refund order 882")
print("   882: yes at 5h, scheduler not run yet ->", ib.answer_approval(db, "refund-882", "yes", later + 5 * h))
ib.request_approval(db, "refund-883", later, 4 * h, "refund order 883")
print("   883: yes at 1h ->", ib.answer_approval(db, "refund-883", "yes", later + h),
      " second tap no at 2h ->", ib.answer_approval(db, "refund-883", "no", later + 2 * h))
ib.run_scheduler(db, later + 5 * h)
print("   883 after its expiry timer fired:", json.loads(db.execute(
    "SELECT payload FROM inbox WHERE source = 'approval' AND source_id = 'refund-883'")
    .fetchone()[0])["answer"])
print("   never requested ->", ib.answer_approval(db, "refund-999", "yes", later))
 
print("8. outbox: three updates inside an hour")
sent = []
m = later + 6 * h
db.execute("UPDATE outbox SET sent_at = ? WHERE sent_at IS NULL", (m - 2 * h,))
for n, minute in enumerate((0, 10, 20)):
    db.execute("INSERT INTO outbox (body, created) VALUES (?, ?)", (f"update {n}", m + minute * 60))
    ib.dispatch(db, m + minute * 60, sent.append)
ib.dispatch(db, m + 61 * 60, sent.append)
print("   sends:", [s.replace("\n\n", " | ") for s in sent])