Lesson 02 of 04 · published

The Report Outbox

~16 min · master-chief, outbox, idempotency, asyncio, sqlite, durability

Level 0Visitor

0 XP0/33 lessons0/11 achievements

0/100 XP to next level100 XP to go0% complete

Three Senders, One Busy Room

Three kinds of report reach Master Chief, and they all go through one function, master_chief.report(). A cron outcome arrives from the scheduler when a job needs Dad (the previous lesson). A follow-up arrives when a check Pippa armed before ending her turn sees delegated work succeed or block (Track 5). A Coop notice arrives through Coop's notify door when delegated work reports back in plain words. All three have to land in a room that is often busy: a long tool turn can hold the Chief's turn lock for minutes, and a report must never cut into a turn halfway.

The First Version Waited

The obvious design is to treat a report like a tiny turn: take the lock, append the row, release. The first version did exactly that, inside the caller, with no bound on the wait. The 2026-10-08 cold review (finding #6) confirmed by probe that the wait had no bound, and traced three separate failures that one line made possible. It graded the chain plausible, not yet observed: the Coop door was under a day old, so nobody had seen it happen.

Coop's notify door could ring again and again. Coop's client gives up after 15 seconds and retries. Each retry would add another waiter on the lock, and once the Chief was free, each waiter would go on to send its own Telegram message.
A cron job could lose its next fire. A job waiting for a long Chief turn would stay "running", and the scheduler, which runs one instance of a job at a time, would skip the next scheduled run.
A restart could lose the report. The text existed only in a coroutine's memory, waiting. Nothing had written it down.

Every one of those would look fine in a test where the room was idle. That is why it took a separate reviewer, reading the code cold, to see it.

Durable Before Anything Waits

The fix (5536821e) changes the order of two things. report() first refuses a source conversation that is not Dad's live Pippa row, then resolves the Chief as its owner (Lesson 4 explains why), then builds an idempotency key, chief-report: plus the caller's source id. If a message with that key already sits in the Chief, it returns that message at once. Otherwise it writes the report into chief_report_outbox with INSERT OR IGNORE, keyed by that same source_key, and commits. Only then does anything wait. A retry that arrives a second later finds either the landed row or the waiting outbox row, and adds nothing.

One Drainer per Chief

Callers no longer append anything themselves. Each call starts the Chief's single drainer task if none is running, or marks it dirty if one is. The drainer takes the Chief's ordinary FIFO turn lock, so it queues behind Dad's turns exactly like a turn would, and when its turn comes it appends every pending outbox row in order: an assistant row whose parent is the newest message, with origin_surface: scheduled-task in its host context, the no-tools sentinel [] so no later read goes searching the log for tools a report never ran, and a turn_kind taken from the source key. It writes the matching JSONL line, marks the outbox row delivered, and before exiting checks the dirty mark, with no await between that check and its return, so a report that arrived mid-drain is never stranded. Stop withdraws Dad's waiting turns but never a report. At startup the main worker calls resume_reports(), which starts a drainer for every Chief with undelivered rows.

Callers Bound Their Wait

Caller	Waits at most	Then
A cron outcome	5 s	The job finishes on time; the line lands when the Chief is free.
Coop's notify door	2 s	Sends a Telegram line unless `coop_notify_sent` already holds the Coop outbox row; a per-row lock keeps concurrent retries to one send. The row is recorded only after the send, so a crash between the two can ring twice. A refused Telegram answers 503 and Coop tries later.
A follow-up	2 s	A blocker has already gone to Telegram first, before the possibly busy lock; the Chief copy lands later.

The wait is wrapped in asyncio.shield, so a caller giving up never cancels the drainer it was waiting on. Giving up is safe because the row was durable before the wait began.

No Reset, No Interruption

One thing a report does not do: touch the turn in progress or the brain's session. Early on, each report cleared the Chief's stored session id so a stateful provider could not skip it with a resume shortcut, which forced a full replay on every report. Dad's ruling on coop #345 asked for that to stop, and d0eaaf16 removed it: every brain already builds each turn by replaying the stored conversation (live rows plus the summary that replaced the oldest), so the next turn includes the report by construction. If you read an older doc line saying reports invalidate the resume, the code moved past it.

A message that must arrive is written down before anyone waits for it. Waiting is where processes die, clients time out and retries multiply. If the record exists first, every one of those becomes harmless.

Code

Engine excerpt (verbatim, the body of report() after its docstring): durable first, then a bounded, shielded wait·python

    store = store or ConversationStore()
    await store.init_db()
    if source_conversation_id:
        source = await store.get_conversation(source_conversation_id)
        if not source or source.owner_user_id != "dad" or source.soul_id != "pippa" or source.trashed_at:
            return None
    chief = await resolve(store, as_owner=True)
    key = "chief-report:" + source_id
    landed = await _landed(store, chief.id, key)
    if landed:
        return landed
    async with aiosqlite.connect(store.db_path, timeout=5) as db:
        await db.execute(_OUTBOX_DDL)
        await db.execute(
            "INSERT OR IGNORE INTO chief_report_outbox (source_key, chief_id, text, source_conversation_id, created_at) "
            "VALUES (?, ?, ?, ?, datetime('now'))",
            (key, chief.id, text, source_conversation_id),
        )
        await db.commit()
    drainer = _drain_reports(store, chief.id)
    try:
        if wait is None:
            await asyncio.shield(drainer)
        else:
            await asyncio.wait_for(asyncio.shield(drainer), timeout=max(wait, 0.0))
    except asyncio.TimeoutError:
        return None
    return await _landed(store, chief.id, key)

A minimal sketch you can run (Python 3.8+, stdlib only, in memory): a report sent and retried during a long turn lands exactly once, after the turn. It shows the order, not a restart; surviving one also takes a file, CREATE TABLE IF NOT EXISTS and a startup drain like resume_reports()·python

import asyncio, sqlite3

db = sqlite3.connect(":memory:")
db.executescript("""
CREATE TABLE messages (id INTEGER PRIMARY KEY, role TEXT, text TEXT, source_key TEXT UNIQUE);
CREATE TABLE report_outbox (source_key TEXT PRIMARY KEY, text TEXT NOT NULL, delivered_at TEXT);
""")
turn_lock = None                    # the lock a conversation turn holds
drainer = None                      # the one drainer for this conversation
dirty = False


def landed(key):
    row = db.execute("SELECT id FROM messages WHERE source_key=?", (key,)).fetchone()
    return row[0] if row else None


async def drain():
    global dirty
    while True:
        dirty = False
        async with turn_lock:       # wait like a turn, never inside one
            pending = db.execute("SELECT source_key, text FROM report_outbox "
                                 "WHERE delivered_at IS NULL ORDER BY rowid").fetchall()
            for key, text in pending:
                if landed(key) is None:
                    db.execute("INSERT INTO messages (role, text, source_key) VALUES ('assistant', ?, ?)",
                               (text, key))
                db.execute("UPDATE report_outbox SET delivered_at=datetime('now') WHERE source_key=?", (key,))
            db.commit()
        if not dirty:               # no await between this check and return
            return


def start_drainer():
    global drainer, dirty
    dirty = True
    if drainer is None or drainer.done():
        drainer = asyncio.create_task(drain())
    return drainer


async def report(text, source_key, wait=None):
    if (mid := landed(source_key)) is not None:
        return mid                  # a retry of a landed report is a no-op
    db.execute("INSERT OR IGNORE INTO report_outbox (source_key, text) VALUES (?, ?)", (source_key, text))
    db.commit()                     # durable BEFORE anything waits
    task = start_drainer()
    try:
        await asyncio.wait_for(asyncio.shield(task), timeout=wait)
    except asyncio.TimeoutError:
        return None                 # the caller moves on; the row lands later
    return landed(source_key)


async def main():
    global turn_lock
    turn_lock = asyncio.Lock()
    async with turn_lock:           # a long turn holds the conversation
        first = await report("backup failed: disk full", "cron:backup:1", wait=0.1)
        retry = await report("backup failed: disk full", "cron:backup:1", wait=0.1)
        waiting = db.execute("SELECT COUNT(*) FROM report_outbox WHERE delivered_at IS NULL").fetchone()[0]
        print("while the turn runs:", first, retry, "waiting in outbox:", waiting)
        await asyncio.sleep(0.2)
    await drainer
    again = await report("backup failed: disk full", "cron:backup:1", wait=1)
    rows = db.execute("SELECT id, text FROM messages").fetchall()
    print("after the turn:", again, rows)


asyncio.run(main())

# Output:
# while the turn runs: None None waiting in outbox: 1
# after the turn: 1 [(1, 'backup failed: disk full')]

External links

Exercise

Find a place in your own code where one part of the system tells another part something important and then waits: a webhook handler that posts to Slack, a job that writes a summary into a shared log, a worker that notifies a user. Ask three questions. If the receiver is busy for five minutes, what does the caller do? If the caller times out and retries, how many messages does the user get? If the process restarts during the wait, where is the message? Then rewrite it so the message is committed to a table keyed by a stable source id before the wait, and a single drainer delivers it.

Hint

Pick the key before you write the table: it must be the same on every retry of the same event, which rules out timestamps and random ids generated at send time. The upstream system's own record id (an outbox row, a job run id, a message id) is usually the right one.

The Report Outbox

Three Senders, One Busy Room

The First Version Waited

Durable Before Anything Waits

One Drainer per Chief

Callers Bound Their Wait

No Reset, No Interruption

Code

External links

Exercise

Progress

Comments 0