SqliteStore — one machine, zero infrastructure¶

SqliteStore is the durable ExecutionStore backend for the single-host case: one process (or several processes on the same box / a shared volume) running statecharts that must survive a restart with no external service to operate. It is the simplest durable backend and the golden exemplar for the SQL family — PostgresStore and RqliteStore are the same shape over a networked engine. If you only need throwaway state, point it at :memory:; if you need persistence, point it at a file path.

What it is and how it stores things¶

Engine: the Python stdlib sqlite3 module — no driver to install, no server to run. The connection is opened once in __init__ and reused for every operation: sqlite3.connect(str(path), check_same_thread=False) (the check_same_thread=False lets the same connection be used from a worker thread, e.g. the timer sweep).
Storage model: each Execution is stored as one row in executions, keyed by its id, with the whole object serialized to JSON in the data column (exe.model_dump_json()). Reloading rehydrates it with Execution.model_validate_json(...). A fresh SqliteStore opened on the same file reads the committed rows, so a run resumes after a process restart.
One write per event is one transaction. The central method, commit, performs the CAS write of the Execution and every side-effect of that step (outbox events, the dedupe marker, timer arms/disarms, spawn intents, the optional trace step) and then calls self._conn.commit() once. Either all of it lands or none of it does — that all-or-nothing property is what makes a crash safe (the state and everything the engine deferred are a single durable unit). See durability.
PRAGMAs. Two are set at open time:
```
PRAGMA journal_mode=WAL    -- readers don't block the single writer
PRAGMA busy_timeout=5000   -- wait up to 5s for the write-lock instead of erroring SQLITE_BUSY
```
WAL (write-ahead logging) lets the monitor’s list_executions/load read while a writer holds the write-lock; busy_timeout makes a contended writer wait for the lock rather than immediately raising database is locked.

Schema¶

Six tables, all created with CREATE TABLE IF NOT EXISTS in __init__ and then committed.

`executions` — the durable state¶

CREATE TABLE IF NOT EXISTS executions
(id TEXT PRIMARY KEY, definition_id TEXT NOT NULL, data TEXT NOT NULL,
 version INTEGER NOT NULL)

Column	Type	Meaning
`id`	TEXT, PRIMARY KEY	the `Execution.id` (a uuid hex). One row per Execution.
`definition_id`	TEXT NOT NULL	the id of the `Definition` this Execution runs — a broken-out column so `list_executions` can filter by it without parsing the blob.
`data`	TEXT NOT NULL	the full Execution serialized, `exe.model_dump_json()`. Holds everything: `status`, `outcome`, `error`, `active_path`, `history`, `context`, `children` (the join counter), `parent_id`/`child_id`, `invoke_seq`, `definition_fqn`, and `version`. The store treats it as an opaque blob except where it reaches inside with `json_extract` for the summary projection.
`version`	INTEGER NOT NULL	the optimistic-concurrency token — a broken-out copy of `Execution.version`. A write succeeds only if the stored `version` still matches the one the Execution was loaded at; the CAS uses this column (not the JSON) so the comparison is a cheap indexed scalar.

data is the source of truth for the Execution; definition_id and version are denormalized out of it purely so the hot paths (CAS, listing) don’t have to parse JSON.

`outbox` — the transactional event outbox¶

CREATE TABLE IF NOT EXISTS outbox
(seq INTEGER PRIMARY KEY AUTOINCREMENT, target_id TEXT, event TEXT NOT NULL)

Column	Type	Meaning
`seq`	INTEGER, PRIMARY KEY AUTOINCREMENT	monotonic delivery order; the ack key.
`target_id`	TEXT (nullable)	the Execution to deliver the event to (`NULL` = no specific target).
`event`	TEXT NOT NULL	the `Event` serialized (`event.model_dump_json()`).

Events emitted by a step are not delivered inline; they are inserted here in the same transaction as the state advance, and a relay drains them after commit (pending_outbox / ack_outbox). A crash between the state write and the delivery cannot lose an emitted Finished — it is already durable in this table.

`processed_events` — at-least-once dedupe¶

CREATE TABLE IF NOT EXISTS processed_events
(execution_id TEXT NOT NULL, event_id TEXT NOT NULL,
 PRIMARY KEY (execution_id, event_id))

Column	Type	Meaning
`execution_id`	TEXT NOT NULL	the Execution that handled the event.
`event_id`	TEXT NOT NULL	the `Event.id` that was processed.
—	PRIMARY KEY (execution_id, event_id)	one marker per (execution, event).

Under at-least-once delivery the same event can arrive twice; recording its id here (and checking it on load) makes the effect take hold once. The composite primary key is what makes the INSERT OR IGNORE in commit a no-op on a redelivery.

`timers` — durable timers¶

CREATE TABLE IF NOT EXISTS timers
(execution_id TEXT NOT NULL, path TEXT NOT NULL, fire_at REAL NOT NULL,
 PRIMARY KEY (execution_id, path))

Column	Type	Meaning
`execution_id`	TEXT NOT NULL	the Execution that armed the timer.
`path`	TEXT NOT NULL	the `full_path` of the state that armed it (a `timeout:` state).
`fire_at`	REAL NOT NULL	wall-clock epoch seconds at which the `Timeout` is due.
—	PRIMARY KEY (execution_id, path)	one live timer per state — re-arming the same path replaces it (the upsert in `commit`).

A timeout: state arms a timer on enter and cancels it on exit, both applied in the same commit as the transition that entered/exited it. The sweep (due_timers) finds rows whose fire_at <= now and publishes a Timeout event.

`spawns` — the orthogonal-fork spawn outbox¶

CREATE TABLE IF NOT EXISTS spawns
(seq INTEGER PRIMARY KEY AUTOINCREMENT, parent_id TEXT NOT NULL, child_id TEXT NOT NULL,
 root_path TEXT NOT NULL, context TEXT NOT NULL)

Column	Type	Meaning
`seq`	INTEGER, PRIMARY KEY AUTOINCREMENT	monotonic order; the ack key.
`parent_id`	TEXT NOT NULL	the orthogonal parent Execution forking the child.
`child_id`	TEXT NOT NULL	the deterministic id of the child Execution to create.
`root_path`	TEXT NOT NULL	the branch node the child runs (its `root_path`).
`context`	TEXT NOT NULL	the child’s seed context, JSON (`json.dumps(context)`).

Entering an AND-state does not create the region children inline; it enqueues one spawn intent per region here, committed atomically with the parent’s advance and its children join expectations. A relay then creates each child idempotently (pending_spawns / ack_spawn). This mirrors the event outbox and is what makes a crash-during-fork safe.

`trace` — the execution-timeline ring (opt-in)¶

CREATE TABLE IF NOT EXISTS trace
(execution_id TEXT NOT NULL, idx INTEGER NOT NULL, entry TEXT NOT NULL,
 PRIMARY KEY (execution_id, idx))

Column	Type	Meaning
`execution_id`	TEXT NOT NULL	the Execution whose timeline this step belongs to.
`idx`	INTEGER NOT NULL	the monotonic step index within that Execution (assigned inline, see below).
`entry`	TEXT NOT NULL	one step (event/transition/actions/context_out), JSON.
—	PRIMARY KEY (execution_id, idx)	one row per step.

Off by default; written by commit only when a trace step is passed. The store keeps a ring of the last trace_max steps per Execution (DEFAULT_TRACE_MAX = 200). Powers the monitor’s timeline without affecting load (which still reads the snapshot, not a replay).

The CAS write (`_write`)¶

_write is the heart of the optimistic-concurrency guarantee. It is deliberately separate from the commit of the transaction so it can be batched with the rest of commit’s statements:

UPDATE executions SET data = ?, version = ? WHERE id = ? AND version = ?

The code:

old = exe.version
exe.version = old + 1
data = exe.model_dump_json()
cur = self._conn.execute(
    "UPDATE executions SET data = ?, version = ? WHERE id = ? AND version = ?",
    (data, exe.version, exe.id, old),
)
if cur.rowcount == 0:
    found = self._conn.execute("SELECT version FROM executions WHERE id = ?", (exe.id,)).fetchone()
    if found is None and old == 0:
        self._conn.execute(
            "INSERT INTO executions (id, definition_id, data, version) VALUES (?, ?, ?, ?)",
            (exe.id, exe.definition_id, data, exe.version),
        )
    else:
        exe.version = old  # undo the in-memory bump; the commit did not happen
        raise StoreConflict(exe.id, expected=old, found=found[0] if found else None)

Step by step:

Capture old = exe.version, then bump exe.version = old + 1. The bump happens before the dump so the JSON in data carries the new version too (the blob and the broken-out version column stay consistent — the blob’s own version field equals the column).
The CAS: the UPDATE ... WHERE id = ? AND version = ? matches the row only if its stored version is still old. If another writer advanced the row in the meantime, the WHERE matches nothing and rowcount == 0.
rowcount == 0 disambiguation. Zero matched rows means one of two things, distinguished by a follow-up SELECT version:
- Brand-new Execution (found is None and old == 0): no row existed yet and we are at the initial version — so this is the first save. Do the INSERT.
- Stale write (a row exists but moved past old, or it’s missing with old != 0): another writer won. Roll back the in-memory bump (exe.version = old, since this transaction will not commit) and raise StoreConflict(exe.id, expected=old, found=...).
_write does not commit. It only issues statements on the connection. That is the whole point: commit can run _write plus the outbox/dedupe/timer/spawn/trace statements and then issue one self._conn.commit(), so they are atomic together. StoreConflict is the single-writer-per-Execution backstop — the caller (a runner/worker) reloads and retries, or drops the stale work.

save is the standalone version used outside an event step: _write then self._conn.commit(), with except StoreConflict: self._conn.rollback(); raise.

`commit` — the one atomic transaction¶

commit is the only write path on the hot loop. It issues a sequence of statements on the open connection and commits them as one transaction, so the state advance and every deferred effect of that event are a single durable unit:

def commit(self, exe, emits, processed_event_id=None, timers=(), spawns=(), trace=None):
    try:
        self._write(exe)                                   # 1. CAS the Execution (no commit yet)

        for target_id, event in emits:                     # 2. enqueue emitted events (outbox)
            self._conn.execute(
                "INSERT INTO outbox (target_id, event) VALUES (?, ?)",
                (target_id, event.model_dump_json()),
            )

        if processed_event_id is not None:                 # 3. record this event handled (dedupe)
            self._conn.execute(
                "INSERT OR IGNORE INTO processed_events (execution_id, event_id) VALUES (?, ?)",
                (exe.id, processed_event_id),
            )

        for child_id, root_path, context in spawns:        # 4. enqueue orthogonal child creations
            self._conn.execute(
                "INSERT INTO spawns (parent_id, child_id, root_path, context) VALUES (?, ?, ?, ?)",
                (exe.id, child_id, root_path, json.dumps(context)),
            )

        for op in timers:                                  # 5. arm / disarm durable timers
            if op.action == "schedule":
                self._conn.execute(
                    "INSERT INTO timers (execution_id, path, fire_at) VALUES (?, ?, ?) "
                    "ON CONFLICT(execution_id, path) DO UPDATE SET fire_at = excluded.fire_at",
                    (exe.id, op.path, op.fire_at),
                )
            else:
                self._conn.execute(
                    "DELETE FROM timers WHERE execution_id = ? AND path = ?", (exe.id, op.path)
                )

        if trace is not None:                              # 6. append a trace step (opt-in)
            self._write_trace(exe.id, trace)

        self._conn.commit()                                # 7. ONE commit: all-or-nothing
    except StoreConflict:
        self._conn.rollback()                              # CAS lost: discard the whole batch
        raise

Statement by statement:

_write(exe) — the CAS UPDATE (or INSERT for a brand-new Execution). If it raises StoreConflict, no further statements run; the except rolls back and re-raises.
Outbox INSERTs — one row per emitted event. Deferred delivery, durable before the relay touches it.
INSERT OR IGNORE into processed_events — mark the event that drove this step as handled. OR IGNORE makes a redelivery a no-op against the composite PK, so dedupe is idempotent.
Spawn INSERTs — one row per orthogonal child to create. Committed together with the parent’s children join expectations (which live inside exe’s data), so the fork and its bookkeeping are atomic.
Timer ops — a schedule is an upsert: ON CONFLICT(execution_id, path) DO UPDATE SET fire_at = excluded.fire_at, so re-arming the same state replaces the prior fire time (the (execution_id, path) PK guarantees one live timer per state). A cancel is a DELETE by (execution_id, path). Both are part of the same transaction as the transition that armed or disarmed them — a scheduled timer can never be lost relative to the state that scheduled it.
_write_trace — append the opt-in trace step (see next section). Only when trace is given.
The single self._conn.commit() — flushes the whole batch atomically.

The except StoreConflict: self._conn.rollback(); raise is the only failure branch: if the CAS lost, the entire batch (state, outbox, dedupe, timers, spawns, trace) is discarded together. Because all of this is one transaction, a crash at any point leaves the store either fully at the prior step or fully at the new one — never half-applied (e.g. a state advanced but its Finished lost, or a fork’s children created but the parent’s join expectations missing). This is the durability contract; the relay and sweep below complete the at-least-once delivery after commit.

The trace ring (`_write_trace`)¶

_write_trace appends one timeline step without committing (so it batches into commit’s transaction). It is two statements:

INSERT INTO trace (execution_id, idx, entry)
SELECT ?, COALESCE((SELECT MAX(idx) FROM trace WHERE execution_id = ?), -1) + 1, ?

then, if trace_max is set, the ring cap:

DELETE FROM trace WHERE execution_id = ? AND idx <=
(SELECT MAX(idx) FROM trace WHERE execution_id = ?) - trace_max

Why it is shaped this way:

idx is computed inline with COALESCE(MAX(idx), -1) + 1 rather than read first and passed in. There is no pre-read round-trip, and the value is monotonic per Execution. Crucially it is computed off MAX(idx), not off a row count — so it keeps climbing even after the ring DELETE removes old rows (a count-based index would collide with a surviving row after eviction). The first step gets idx = 0 (COALESCE(NULL, -1) + 1).
The cap deletes everything whose idx is at least trace_max behind the current max, keeping only the last trace_max steps. With DEFAULT_TRACE_MAX = 200 the timeline is a 200-step ring per Execution.

read_trace re-stamps the index from the idx column, so the stored JSON entry need not carry its own index:

SELECT idx, entry FROM trace WHERE execution_id = ? ORDER BY idx
-> [{**json.loads(entry), "index": idx} for idx, entry in rows]

append_trace is the demo/test seam — it writes a step and commits it on its own (used to seed a timeline outside a real event step):

def append_trace(self, execution_id, entry):
    self._write_trace(execution_id, entry)
    self._conn.commit()

Reads and sweeps¶

`load`¶

SELECT data FROM executions WHERE id = ?

Returns Execution.model_validate_json(row[0]) or None if no row. The full snapshot, rehydrated.

`load_for_event` (async only)¶

The sync SqliteStore has no load_for_event; the worker uses load + is_processed separately. The async twin folds both into one round-trip (see Async twin below). The ExecutionStore Protocol lists load_for_event as optional sugar.

`list_executions`¶

The monitor/list view. It never selects the heavy data blob; it projects only the scalar summary fields, reaching inside the JSON with json_extract for the ones that aren’t broken-out columns:

SELECT id, definition_id, version, json_extract(data,'$.status'),
       json_extract(data,'$.outcome'), json_extract(data,'$.active_path'),
       json_extract(data,'$.parent_id') FROM executions
WHERE <conditions> ORDER BY id LIMIT ? OFFSET ?

WHERE building. Conditions start at ["1=1"] and accumulate: definition_id = ? (exact match, a real column); json_extract(data,'$.status') IN (?, ?, ...) for a status set (OR); json_extract(data,'$.parent_id') IS NULL when roots_only. Because status/outcome/ parent_id live inside the blob, json_extract is how the filters reach them — no separate columns needed.
Pagination. The cursor is an opaque base64 of an integer offset (_decode_offset / _encode_offset). It fetches limit + 1 rows: if more than limit come back there is a next page, so it returns the first limit as ExecutionSummary items and a next_cursor encoding off + limit; otherwise next_cursor is None. Ordering is stable by id.

`is_processed`¶

SELECT 1 FROM processed_events WHERE execution_id = ? AND event_id = ?

Returns True if the marker exists — the dedupe lookup under at-least-once delivery.

`pending_outbox` / `ack_outbox`¶

SELECT seq, target_id, event FROM outbox ORDER BY seq          -- oldest first
DELETE FROM outbox WHERE seq = ?                               -- then commit

pending_outbox returns the undelivered OutboxEntry list (oldest first), deserializing each event. ack_outbox(seq) removes a delivered entry and commits on its own (the relay’s ack is its own small transaction, independent of the event step that enqueued it).

`pending_spawns` / `ack_spawn`¶

SELECT seq, parent_id, child_id, root_path, context FROM spawns ORDER BY seq
DELETE FROM spawns WHERE seq = ?                               -- then commit

pending_spawns returns SpawnEntry rows (oldest first), json.loads-ing the context. ack_spawn(seq) removes a created child’s intent and commits. The relay creates each child idempotently before acking, so a crash between create and ack just retries a harmless create.

`due_timers` / `delete_timer`¶

SELECT execution_id, path, fire_at FROM timers WHERE fire_at <= ? ORDER BY fire_at

due_timers(now) returns the timers whose fire_at <= now as (execution_id, path, fire_at), soonest first — the sweep publishes a Timeout for each.

DELETE FROM timers WHERE execution_id = ? AND path = ? AND fire_at = ?

delete_timer is guarded on fire_at: it deletes the row only if it still holds the exact fire time the sweep saw. This is the subtle, important bit — if the model re-scheduled the same (execution_id, path) to a new time between the sweep reading it and acting on it, the fire_at no longer matches, the DELETE removes nothing, and the freshly-armed timer survives the stale sweep instead of being silently dropped. It commits on its own.

`close`¶

self._conn.close() — releases the connection.

Async twin (`aio_store/sqlite.py`)¶

AsyncSqliteStore is a faithful mirror of SqliteStore over aiosqlite — same schema, same SQL, same CAS, same one-transaction commit. The differences are mechanical:

Every DB call is awaited (await self._conn.execute(...), await cur.fetchone(), await self._conn.commit()).
Atomicity still holds. aiosqlite serializes a connection’s operations on its own worker thread, so the multi-statement commit runs as one ordered batch on that thread and remains a single atomic transaction — exactly like the sync version. The awaits interleave with other tasks at the event loop, not mid-transaction.
Construction is async: the connection must be awaited open, so you build it with await AsyncSqliteStore.create(path) (a classmethod that opens the connection, sets the PRAGMAs, creates the tables, commits, and returns the instance). __init__ just stores the connection and sets self.trace_max = DEFAULT_TRACE_MAX (the sync version sets it after the CREATE TABLEs; the async version sets it in __init__).
It additionally provides load_for_event — load + dedupe-check in one round-trip (the worker’s per-event pair), a single query that selects the data and an EXISTS over processed_events together:
```
SELECT (SELECT data FROM executions WHERE id = ?),
       EXISTS(SELECT 1 FROM processed_events WHERE execution_id = ? AND event_id = ?)
```
It returns (Execution | None, already_processed: bool); if the data sub-select is NULL (no such Execution) it returns (None, False).

:memory: works in both as a non-persistent variant for tests.

When to pick it / tradeoffs¶

Pick SqliteStore when:

You run on one host (or several processes over a shared volume) and want durability with zero infrastructure — no server, no driver to install.
A single writer per Execution is acceptable: WAL gives concurrent readers, but SQLite has one writer at a time, and the version CAS is the single-writer-per-Execution backstop. This is exactly the model the engine assumes.
You want the simplest thing that survives a restart, or :memory: for fast tests.

Reach for a networked backend instead when you need many hosts writing concurrently (horizontal scale, no shared filesystem): PostgresStore (same SQL shape, server-side CAS via UPDATE ... WHERE version) or RqliteStore (distributed SQLite over Raft). Those are the same design as this page, just over a network engine — SqliteStore is the reference to read first.

See the stores hub for the full backend matrix and the STM_STORE_BACKEND selector, and durability for the crash-safety guarantees this backend implements.

SqliteStore — one machine, zero infrastructure¶

What it is and how it stores things¶

Schema¶

executions — the durable state¶

outbox — the transactional event outbox¶

processed_events — at-least-once dedupe¶

timers — durable timers¶

spawns — the orthogonal-fork spawn outbox¶

trace — the execution-timeline ring (opt-in)¶

The CAS write (_write)¶

commit — the one atomic transaction¶

The trace ring (_write_trace)¶

Reads and sweeps¶

load¶

load_for_event (async only)¶

list_executions¶

is_processed¶

pending_outbox / ack_outbox¶

pending_spawns / ack_spawn¶

due_timers / delete_timer¶

close¶

Async twin (aio_store/sqlite.py)¶