Anthropics Leitfaden zu Always-on-Agenten: was Inbox, Timer und Outbox tun, wenn ein Worker stirbt, getestet in SQLite

Anthropic hat am 13. November „Building always-on agents“ veröffentlicht. Ein Always-on-Agent schläft laut dem Leitfaden die meiste Zeit und wacht für kurze Runs auf, und die Schleife innerhalb eines Runs ist dieselbe, die jeder Agent nutzt. Was sich ändert, ist die Maschinerie drumherum: eine Event-Inbox, Timer, Memory, Berechtigungen und eine Outbox. (Zu den Argumenten des Leitfadens dafür, das jetzt anzugehen, gehört eine Kostenzahl: Auf einem Long-Horizon-Coding-Benchmark, den er nicht nennt, kostete Opus 5.5 pro gelöster Aufgabe 78% weniger als Opus 5, gegenüber 40% weniger bei typischen Workloads. Ein Benchmark des Herstellers, also eine Richtung, kein Preis.)

Der Leitfaden benennt die Teile gut und hört vor den Fehlerfällen auf, also genau dort, wo ein unbeaufsichtigter Agent seine interessanten Nächte verbringt. Ein Agent-Run ist ein teurer Job mit Seiteneffekten: Ein Duplikat ist eine zweite E-Mail an einen Kunden, keine verschwendete CPU-Sekunde. Deshalb wiegen die Lücken hier schwerer als in einer Job-Queue, die Bilder skaliert.

Die meisten davon schließt ein einziger Schritt. Mach jedes Aufwachen zu einem Inbox-Event, Timer-Ticks, Antworten auf Freigaben und abgelaufene Freigaben eingeschlossen, gib jedem einen Schlüssel aus seiner Quelle und lass eine UNIQUE (source, source_id)-Constraint die zweite Kopie ablehnen. Der Rest (wer den Lock hält, was mit einem Event passiert, das jeden Worker abstürzen lässt, wie oft der Agent sich melden darf) braucht eine Lease mit Generationsnummer und ein paar Zählspalten. Unten steht alles als eine Python- und SQLite-Datei, jeder Teil ausgeführt gegen den Fehler, für den er existiert.

Was deckt Anthropics Leitfaden zu Always-on-Agenten ab, und was lässt er offen?

Er deckt die Form ab: Events aus Webhooks, Zeitplänen, Hintergrundjobs und Nutzernachrichten landen alle in einer dauerhaften Inbox, ein einziger Worker pro Inbox führt den Agenten aus, Timer liegen in einem eigenen Speicher und speisen die Inbox, Memory liegt in dauerhaftem Speicher, Berechtigungen entscheiden über jeden Tool-Call, und die Outbox liefert Ergebnisse statt Schritt-für-Schritt-Streams. Offen lässt er, wie das alles hält, wenn etwas ausfällt.

Zwei der fünf Teile haben hier schon eigene Artikel. Memory, samt der Faustregel des Leitfadens, nur zu speichern, was der Agent nicht erneut nachschlagen kann, steht in Anthropics Komponenten des Agent-Memory in Claude Code. Die unbeaufsichtigte Seite der Berechtigungen steht in denyInsteadOfPrompting in Claude Code. In diesem Artikel geht es um Inbox, Timer, Freigaben und Outbox. Die offenen Fragen, so wie ich die Seite lese:

• wie „an inbox should only have one active worker at a time“ durchgesetzt wird und was mit einem Worker passiert, der nur langsam war, als sein Ersatz startete;
• Retries: Stirbt ein Worker mitten im Run, übernimmt ein Ersatz dasselbe Event und versucht es erneut, ohne genanntes Limit;
• ein Event, das jeden Worker abstürzen lässt, der es anfasst;
• was ein Scheduler mit Terminen macht, die er verpasst hat, während er nicht lief;
• was aus einem aufgeschobenen Tool-Call wird, wenn der Nutzer nie antwortet;
• wie oft die Outbox sich melden darf. „Start quiet and then earn more of the user’s attention over time“ ist der richtige Instinkt, aber keine Zahl.

Wie speichert die Inbox eines Always-on-Agenten einen wiederholten Webhook nur einmal?

Gib jedem Event die ID seiner Quelle als Schlüssel und lass die Datenbank die zweite Kopie ablehnen. Der Leitfaden sagt es direkt: „events can be deduplicated based on unique identifiers from the source.“ GitHub schreibt eine Delivery-ID in den Header X-GitHub-Delivery, Stripe gibt jedem Event eine ID. Eine einzige Constraint macht aus einer wiederholten Zustellung ein No-op:

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)
);

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

Der Leitfaden will, dass der Handler „acknowledge events quickly“ und tiefere Prüfungen, Prompt-Injection-Screening eingeschlossen, in den Run verschiebt. Billige Prüfungen kommen trotzdem zuerst: Prüf die Signatur des Anbieters vor ingest(), sonst kann jeder, der den Endpoint erreicht, Events schreiben und Source-IDs belegen, bevor die echten Zustellungen ankommen. Danach schnell bestätigen, denn ein Absender, der beim Warten auf dein 200 in einen Timeout läuft, stellt unter Umständen erneut zu; Stripe wiederholt fehlgeschlagene Live-Mode-Zustellungen bis zu drei Tage lang.

Einmal gespeichert ist nicht gleich einmal ausgeführt. Zwei verschiedene Events, die dasselbe bedeuten („deploy it“ zweimal geschickt), sind zwei Schlüssel, und das Batching des Leitfadens gruppiert sie nur: Der Worker nimmt alle wartenden Events auf einmal, sodass aus drei schnellen Nachrichten ein Run und eine Antwort werden. In meinem Lauf kamen drei direkt hintereinander eingegangene Chat-Nachrichten als ein Batch von drei heraus. Ob der Agent einmal deployt, liegt trotzdem beim Agenten und beim Deploy-Tool.

Wie erzwingst du bei einem Always-on-Agenten einen Worker pro Inbox?

Mit einer Lease, die eine Generationsnummer trägt. Die Lease-Zeile sagt, wem die Inbox gehört und bis wann; jede Übernahme erhöht die Generation; jedes geclaimte Event wird mit der Generation gestempelt, die es geclaimt hat; und der Schreibvorgang, der einen Run abschließt, prüft Holder, Generation und Event-Status in einer Transaktion. Der Leitfaden empfiehlt eine Inbox pro Nutzer. Hier ist das eine SQLite-Datei pro Nutzer, und die Lease-Tabelle hat genau eine Zeile.

def acquire(db, holder, now):
    """Take or renew the lease. Returns the generation, or None if someone else holds it."""
    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 finish(db, holder, gen, now, ids, messages=(), error=None):
    """Record a run's outcome and queue its messages in one fenced transaction."""
    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

Eine Lease läuft nach der Uhr ab, und ein Worker, der nur langsam ist (ein langer Tool-Call, ein zugeklappter Laptop), merkt nicht, dass seine Zeit um ist. Nur den Holder-Namen zu prüfen reicht nicht. In meinem Szenario hat Worker A ein Event geclaimt und ist hängen geblieben. B hat übernommen, nachdem die 120-Sekunden-Lease abgelaufen war (Generation 2), hat das Event abgeschlossen, seine Nachricht eingereiht und ist dann in den Leerlauf gegangen. Als A aufwachte, waren beide Leases verfallen, also konnte A die Lease ganz regulär erneut nehmen, als Generation 3. Dann versuchte A, seinen alten Claim aus Generation 1 abzuschließen, und finish lehnte ab: Das Event blieb done, mit einer Nachricht in der Outbox, nicht zwei. Mit einer reinen Namensprüfung geht dieses verspätete Finish durch.

Die Nachricht kommt in derselben Transaktion in die Outbox, in der das Event abgeschlossen wird. So kann ein Absturz weder ein abgeschlossenes Event mit verlorener Antwort hinterlassen noch eine eingereihte Antwort zu einem Event, das noch einmal laufen wird.

Was die Prüfung nicht rückgängig machen kann, ist ein Seiteneffekt, den A vor dem Hängenbleiben ausgelöst hat. Hatte A die E-Mail schon selbst verschickt, verschickt B sie noch einmal. Ein Retry nach einer Übernahme ist At-least-once-Ausführung, genau der Grenzfall, über den ich in Multi-Agenten-Systeme als verteilte Systeme geschrieben habe: Jedes Tool, das ein unbeaufsichtigter Run aufrufen kann, muss einen doppelten Aufruf vertragen oder einen Schlüssel mitgeben, über den das empfangende System dedupliziert. Setz die Lease deutlich über deinen längsten normalen Run und erneuere sie zwischen den Schritten, sonst wird aus jedem langen Run ein doppelter.

Mein Test „zwei Worker im selben Moment“ bestand aus zwei sequenziellen Aufrufen mit demselben Zeitstempel (A bekam Generation 1, B bekam None), er zeigt also die Regel, kein Race. Auf einem Host sorgt BEGIN IMMEDIATE dafür, dass SQLite die Schreiber serialisiert.

Was passiert mit dem Event eines Always-on-Agenten, wenn der Worker mittendrin stirbt?

Es geht zurück in die Queue, nur begrenzt oft und für sich allein. Die Recovery-Geschichte des Leitfadens lautet, dass ein Ersatz-Worker dasselbe Event übernimmt und es erneut versucht. Wörtlich genommen wird ein Event, das den Worker abstürzen lässt (ein Payload, der den Kontext sprengt, ein Tool, das daran stirbt), übernommen, lässt den nächsten Worker abstürzen und so weiter, während alles dahinter wartet.

Drei Regeln in claim_batch fangen das ab. Zähl den Versuch beim Claim, nicht beim Fehlschlag: Ein Worker, der abstürzt, verbucht nie einen Fehlschlag. Setz Zeilen, die eine ältere Generation gestempelt hat, auf pending zurück, denn ihr Worker hat die Lease verloren. Und claim ein erneut versuchtes Event allein, damit ein Poison-Event die gesunden Events aus seinem ersten Batch nicht mit in den Abgrund zieht.

    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

Ich habe ein Poison-Event zusammen mit einem gesunden eingereiht, und jeder Worker, der das Poison-Event claimte, beendete sich ohne Finish. W0 nahm beide und stürzte ab. W1 und W2 versuchten das Poison-Event allein und stürzten ab. Danach war es nach drei Versuchen dead, W3 nahm das gesunde Event und schloss es ab, und W4 fand nichts. Das gesunde Event hat überlebt, musste aber hinter drei Abstürzen warten; Retries nach Alter abzuarbeiten tauscht Latenz gegen Einfachheit, und eine stark ausgelastete Inbox würde sie womöglich zwischen andere Events einstreuen. Zwei weitere Grenzen: Die hier gezeigte Version hat kein Backoff zwischen den Versuchen, und sie setzt einen Claim pro Run voraus, der nach dem finish des vorherigen Runs passiert.

Dead-Zeilen bleiben in der Tabelle. Genau sie sollten die tägliche Zusammenfassung oder das Alerting des Agenten lesen, denn ein Poison-Event, das niemand sieht, sieht einfach aus wie ein Agent, der verstummt ist.

Wie sollten die Timer eines Always-on-Agenten mit doppeltem Auslösen oder einem verpassten Tag umgehen?

Rück den Timer in derselben Transaktion vor wie das Insert, bilde den Event-Schlüssel aus Timer und geplanter Zeit und fass verpasste Termine zu einem Event zusammen. Der Leitfaden legt Timer in einen eigenen Speicher, mit einem Scheduler, der „adds an event to the inbox and wakes the worker“; ein Timer „doesn’t initiate a run on its own.“ Erst dass der Tick durch die Inbox läuft, macht ihn deduplizierbar.

        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,))

Die beiden Schutzmechanismen fangen Unterschiedliches ab. Ein zweiter Scheduler-Durchlauf im selben Moment findet nichts Fälliges, weil der erste den Timer schon vorgerückt hat: In meinem Lauf behandelte Durchlauf 1 einen Timer und Durchlauf 2 keinen. Der Schlüssel fängt Replays ab, die die Transaktion nicht sehen kann, etwa ein Deploy, das den Zeitplan mit seiner alten Fälligkeit neu registriert. Ich habe den Timer für denselben Termin neu registriert und den Scheduler laufen lassen: Er fand einen fälligen Timer, und die Inbox enthielt weiterhin ein Event. Zeiten sind ganzzahlige Sekunden, damit der Schlüssel exakt bleibt; ein Float-Zeitstempel kann um einen Rundungsschritt abweichen und für denselben Termin einen zweiten Schlüssel erzeugen.

Nach einem simulierten Ausfall, der vier tägliche Termine abdeckte, erzeugte der Scheduler ein weiteres Event für den letzten Termin, zehn Minuten zu spät, mit skipped: 3 im Payload, und setzte den nächsten Lauf auf 23,8 Stunden später. Ob zusammengefasst wird, entscheidest du pro Timer. Ein täglicher Digest sollte zusammenfassen, denn niemand will um 09:10 vier Digests. Ein täglicher Abgleich der Abrechnung sollte es nicht und bräuchte ein Event pro verpasstem Tag unter eigenem Schlüssel. Wenn du skipped an den Agenten weitergibst, kann er sagen „das umfasst vier Tage“, statt so zu tun, als wäre nichts passiert.

Was passiert, wenn niemand auf die Freigabeanfrage eines Always-on-Agenten antwortet?

Das sagt der Leitfaden nicht. Ein aufgeschobener Tool-Call beendet den Run, „the session is saved mid-call“, und die Antwort des Nutzers kommt als Inbox-Event zurück, das die Session fortsetzt. Ohne Ablaufzeit bleibt ein Ja unbegrenzt gültig, und die Aktion läuft, wann immer jemand darauf tippt, Tage später und ohne Kontext. Dieselbe Lücke zeigt sich bei MCP-Elicitations: Ein unbeaufsichtigter Prozess kann eine Frage stellen, ohne dass jemand da ist, der sie beantwortet.

Die Lösung hat zwei Teile, beide mit Schlüssel wie alles andere. Die Ablaufzeit ist ein einmaliger Timer, der eine expired-Antwort unter dem eigenen Schlüssel der Freigabe schreibt, sodass eine Anfrage, auf die niemand antwortet, die Session trotzdem fortsetzt, als Ablehnung. Und der Antwortpfad prüft die Frist selbst, in derselben Transaktion, statt darauf zu vertrauen, dass der Scheduler schon gelaufen ist:

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"]

Pro Anfrage wird immer nur eine Antwort gespeichert, und die Funktion gibt die zurück, die gilt, damit der Handler dem Nutzer sagen kann, was tatsächlich passiert ist. Ich habe drei Freigaben für Rückerstattungen mit einem Vier-Stunden-Fenster simuliert. Als der Ablauf-Timer bei Stunde fünf ausgelöst hatte und bei Stunde sechs ein Ja kam, war das Ergebnis expired. Ein Ja bei Stunde fünf, bevor der Scheduler gelaufen war, kam ebenfalls als expired zurück; das ist der Fall, den eine Ablaufzeit, die nur im Timer lebt, falsch behandelt. Ein Ja bei Stunde eins galt, ein zweites Tippen auf Nein bei Stunde zwei lieferte yes, und der später auslösende Ablauf-Timer änderte nichts. Eine Antwort auf eine nie gestellte Anfrage lieferte unknown.

Wie drosselst du die Outbox eines Always-on-Agenten?

Halte Nachrichten in einer Tabelle fest und lass einen Dispatcher sie mit einem Mindestabstand senden, wobei er zusammenführt, was sich angestaut hat. Der Leitfaden sagt „you only need a formal outbox when you want to pace your agent’s messages or allow your agent to recover from errors,“ und dass der Agent leise anfangen soll. Ein Mindestabstand ist die kleinste Regel, die diese Zurückhaltung außerhalb des Modells erzwingt:

def dispatch(db, now, send):
    """Send the held messages as one, at most once per OUTBOX_GAP."""
    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)

Drei Updates, eingereiht bei Minute 0, 10 und 20, gingen bei einem Abstand von einer Stunde als zwei Sendungen raus: die erste sofort, die anderen beiden zusammengeführt bei Minute 61. Das Senden passiert innerhalb der Schreibtransaktion, also können zwei Dispatcher nicht beide dieselben Zeilen senden. Es bleibt trotzdem at-least-once: Ein Absturz nach send() und vor dem Commit sendet die zusammengeführte Nachricht noch einmal. Markierst du die Zeilen vor dem Senden als gesendet, wird daraus at-most-once, und welchen Verlust du lieber in Kauf nimmst, hängt vom Kanal ab.

Der Abstand begrenzt nur die Häufigkeit. Der Leitfaden nennt drei Arten, wie ein Run enden kann (eine Nachricht geht raus, nichts geht raus, oder ein aufgeschobener Tool-Call fragt den Nutzer), und die Entscheidung für „nichts“ muss vom Agenten kommen: Ein Run, der keine Änderung gefunden hat, sollte ohne Nachricht enden, und in den Evals, die der Leitfaden empfiehlt, prüfst du, dass er das tut. Dringende Nachrichten brauchen eine explizite Priorität auf der Zeile, die den Abstand umgeht. Sonst entscheidet das Modell, wann der Abstand gilt.

Braucht ein Always-on-Agent für einen einzelnen Nutzer das Credential-Proxy-Muster aus dem Leitfaden?

Das hängt davon ab, welchen Text der Agent liest, nicht davon, wie vielen Nutzern er dient. Das Muster des Leitfadens ist „a proxy outside the agent’s environment that adds credentials to outgoing requests.“ Was es bringt, ist eng umrissen: Das rohe Secret gelangt nie in das, was der Agent sehen kann, also kann es nicht in einem Transkript, einer Memory-Datei oder einem Tool-Argument landen, und eine Anweisung, versteckt in einer E-Mail, die der Agent zusammenfassen sollte, kann es nirgendwohin schicken.

Ein Agent, der nur die Nachrichten und Repositories seines Besitzers verarbeitet, liest wenig nicht vertrauenswürdigen Text, und ein eng gefasstes Token mit kurzer Laufzeit kann genügen. Ein Agent, der ein öffentliches Postfach, Webhooks von Dritten oder Webseiten liest, liest von Angreifern kontrollierten Text, zeitgesteuert und ohne dass jemand zuschaut, und der Leitfaden verlegt das Injection-Screening genau in diesen Run. Für diesen Fall ist der Proxy da. Das Screening wird irgendwann etwas übersehen, und der Proxy begrenzt, was dieser Fehlgriff erreichen kann, vorausgesetzt, der gesamte ausgehende Traffic läuft über ihn, er schränkt die Ziele ein und seine eigenen Logs lassen das Credential weg. Er hindert den Agenten nicht daran, eine Anfrage zu missbrauchen, die er stellen darf; er gibt dir eine Stelle, an der du den Zugriff ab diesem Moment kappen kannst.

Die Schleife eines Always-on-Agenten bleibt gleich, und die Garantien stecken in der Inbox

Jedes Ergebnis oben stammt aus einem Lauf des Drivers am Ende gegen das Skript am Ende. Die Rohausgabe:

1. webhook delivered twice
   first: True  retry: False  rows: 1
2. two workers ask for the lease at the same time
   A: 1  B: None
3. three quick chat messages
   events in one run: 3
4. A stalls mid-run; B takes over; A wakes up
   B before A's lease expires: None
   B after expiry: gen 2  re-claims ['msg-3']
   A renews after both leases lapsed: gen 3
   A finishes its old claim (gen 1 ): False
   msg-3: ('done', 2)  outbox rows: 1
5. a poison event arrives together with a healthy one
   W0 claimed ['poison', 'healthy'] and crashed
   W1 claimed ['poison'] and crashed
   W2 claimed ['poison'] and crashed
   W3 claimed ['healthy'] and finished
   W4 claimed [] and finished
   poison: ('dead', 3)  healthy: ('done', 2)
6. timers: a second scheduler pass, a replay, a three-day outage
   pass 1: 1  pass 2: 0
   replay of the same occurrence: 1 due, 1 event total
   after the outage: events total 2  skipped 3  late by 10 min  next due in 23.8 h
7. approvals with a four-hour window
   881: expiry ran at 5h, yes at 6h -> expired
   882: yes at 5h, scheduler not run yet -> expired
   883: yes at 1h -> yes  second tap no at 2h -> yes
   883 after its expiry timer fired: yes
   never requested -> unknown
8. outbox: three updates inside an hour
   sends: ['update 0', 'update 1 | update 2']

Das ist ein Prozess, eine SQLite-Datei und eine simulierte Uhr. Es zeigt die Logik jeder Regel, nicht, wie die Regeln unter echter Nebenläufigkeit halten. Auf einem Host serialisiert BEGIN IMMEDIATE die Schreiber. Über mehrere Maschinen hinweg braucht das Design eine Server-Datenbank, und die Portierung ist mehr als Suchen und Ersetzen: Postgres hat kein BEGIN IMMEDIATE, also brauchen Lease-Prüfung und Finish explizite Row-Locks; Zeitstempel brauchen einen richtigen Integer- oder Timestamp-Typ; und der übliche Claim für mehrere Consumer, FOR UPDATE SKIP LOCKED, erzwingt für sich allein noch keinen einzelnen aktiven Run pro Inbox. Diese Version habe ich nicht laufen lassen.

Der Agent ist im Driver ein Stub, und nichts davon berührt den Modellaufruf. Der Leitfaden hat recht damit, dass die Schleife gleich bleibt. Was du baust, ist die Buchführung drumherum, und der Großteil dieser Buchführung lief am Ende auf einen einzigen Unique-Key hinaus.

"""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])