Skip to content

Persistence and recovery

By default an Engine keeps everything in memory. Give it a durable store and runs survive restarts; add recover() and a crashed process's open runs are continued.

Choosing a store

Store Durability Processes Use for
InMemoryEventStore none one tests, notebooks, short-lived scripts
SQLiteEventStore(path) disk (WAL) one engine (readers are fine) a service on one machine, local development
PostgresEventStore.create(dsn) database many production, clusters
RedisEventStore.create(url) your Redis persistence settings many low-latency clusters where Redis already runs

All implement the same EventStore protocol and pass the same contract tests, so switching is a one-line change.

from aaa.events import SQLiteEventStore

store = SQLiteEventStore("agent.db")  # creates the file and schema
async with Engine(store=store) as engine:
    ...
await store.aclose()

SQLite specifics

  • Pure standard library (no extra install). One dedicated thread owns the connection, so the event loop never blocks on disk.
  • WAL mode, synchronous=NORMAL: crash-safe (the file cannot be corrupted), but a power loss may lose the last few commits (never reorders or tears them). For strict durability pass SQLiteEventStore(path, synchronous="FULL").
  • Appends are single IMMEDIATE transactions: the optimistic-concurrency check and the inserts are atomic, and a failed append leaves nothing behind.
  • One engine process per database file. The in-process bus has no cross-process coordination; two engines on one file would both react to the same events. For multiple processes use Postgres or Redis.

Recovering after a crash

"""Durable runs: kill the process mid-run, start a new one, and the run finishes where it left off."""

import asyncio
import tempfile
from pathlib import Path

from aaa.algos import ToolLoop
from aaa.events import SQLiteEventStore
from aaa.llm import FakeLLMClient, LLMHandler
from aaa.llm.fake import text, tool_call
from aaa.runtime import Engine
from aaa.tools import ToolHandler, ToolRegistry, tool


@tool
def lookup(key: str) -> str:
    """Look something up."""
    return f"value-of-{key}"


class HangsOnSecondCall:
    """A model whose second call never returns, like a process that dies mid-request."""

    calls = 0

    async def complete(self, request):
        self.calls += 1
        if self.calls == 1:
            return tool_call("lookup", key="x")
        await asyncio.Event().wait()

    async def embed(self, *a, **k): ...
    async def aclose(self): ...


async def boot(db: Path, llm):
    store = SQLiteEventStore(db)
    engine = await Engine(store=store).__aenter__()
    LLMHandler(llm).attach(engine)
    registry = ToolRegistry([lookup])
    ToolHandler(registry).attach(engine)
    return engine, ToolLoop(registry), store


async def main() -> None:
    db = Path(tempfile.mkdtemp()) / "agent.db"

    # --- process 1: starts a run, then "crashes" with the LLM request in flight
    engine, agent, store = await boot(db, HangsOnSecondCall())
    await engine.start(agent, {"task": "look up x"}, run_id="job-42")
    while not any(e.type == "aaa.tool.completed" for e in await engine.store.read("job-42")):
        await asyncio.sleep(0.02)
    await asyncio.sleep(0.2)
    await engine.abort()  # models a crash: nothing is completed, drained, or cleaned up
    await store.aclose()
    print("process 1 died with the run open")

    # --- process 2: same database, new engine; recover() re-adopts every open run
    engine, agent, store = await boot(
        db, FakeLLMClient(lambda r: text(f"result: {r.messages[-1].content}"))
    )
    handles = await engine.recover([agent])
    outcome = await handles["job-42"].result()
    print(f"process 2 recovered {len(handles)} run(s):", outcome.value["text"])
    await engine.__aexit__(None, None, None)
    await store.aclose()


asyncio.run(main())

process 1 died with the run open
process 2 recovered 1 run(s): result: value-of-x

await engine.recover([algo, ...]) returns {run_id: RunHandle} for every run it re-adopted. Pass every algorithm that might have open runs (children included, e.g. the PlanAndExecute and its executor).

What recover() does:

  1. Re-adopts each open run: rebuilds its budget, spend and deadline from the log. A deadline keeps counting through the outage, so a run that should have timed out is completed as timeout immediately.
  2. Repairs relationships: cancels orphaned children, re-announces a finished child's result that never reached its parent.
  3. Redelivers every event that nothing reacted to (an event is "reacted to" if some other event names it as its cause). Because algorithms decide from state and derived ids are deterministic, repeating a reaction is absorbed.

engine.last_recovery is a RecoveryReport: recovered, unknown_algo, orphans_cancelled, timed_out, announced, redelivered.

Open runs whose algorithm you did not pass

They are left untouched and listed in last_recovery.unknown_algo. Recovery needs unique algorithm names (it matches by name) and raises ValueError if two different instances share one.

What happens to work in flight

In flight at the crash After recovery
An LLM request with no response Requested again (you pay for the repeated call).
A tool call that had not started Runs.
A tool call that had started but had no result Depends on its delivery policy: re-run as attempt 2 (at_least_once) or reported as OutcomeUnknown (at_most_once).
A tool call that finished Not repeated: its result is on the log.
A pending approval Still pending; it is just events.
A running child run Resumed with its parent.

Clusters do not call recover(): the distributed bus redelivers by itself. See Running a cluster.

Growth and retention

  • Nothing is deleted unless you configure it. The log is append-only; completed runs stay until you erase them with engine.erase/purge or retention policies, which also remove snapshots, blobs and outbox copies.
  • For the distributed bus, trim the outbox periodically (store.trim_outbox(keep_last=...); see Running a cluster).
  • Blobs of erased runs are deleted with them; see Large payloads.
  • Long conversations are linear in size (Performance).

Exporting events

aaa.events.cloudevents.to_cloudevent(event) yields a CloudEvents 1.0 dict you can ship to Kafka, a warehouse or a tracing pipeline. See Events and the log.