Add a state store for another SQLAlchemy database by subclassing SQLStateStore, plug it in, create its schema, then certify it with the published pytest test kit.

Build your own store

Dendrux keeps every run in a state store. We ship stores for PostgreSQL, SQLite and MySQL/MariaDB. If your database is one SQLAlchemy supports, you can add it by subclassing one base class and filling in a small set of dialect hooks, then prove it with the test kit we publish.

Scope. Stores for relational databases through SQLAlchemy are supported. Stores for other kinds of database (document stores, key-value stores, files) are not part of 0.6.0. The StateStore protocol and the test kit are public, so you can still write one, but we ship and support SQL stores only.

What the base class does for you

SQLStateStore owns every operation: transaction shape, validation, rollback, lock order, retries, sequence numbers, fencing, and the checks that make a run resumable. Your subclass supplies two things:

1. Feature flags that choose between generic code paths in the base.
2. Hook methods that build the SQL expressions or statements the base cannot write portably.

The base also refuses, on first use and before any claim, an engine or connection in AUTOCOMMIT mode (a ValueError): every write must be one real transaction. A hook never commits, retries or validates. It only returns SQL (or, for three of them, runs one statement inside the transaction the base already opened).

Everything a hook takes or returns is importable from dendrux.store: SQLStateStore, ConflictSpec, InsertPlan, SelectSource, ValuesSource, SetFields, AppendArray, PatchLoopState, JsonPatch, JsonPath, JsonReadType, JsonValue, ScalarKey, OrderedAssignments, RunLockScope, RunParentCursor, DialectFacts, DbError, DbErrorKind and TenantScope. Names that start with an underscore are not part of the contract.

A complete store

The skeleton below is a working store. It targets SQLite so that you can run it today and watch the kit pass; for your database, keep the shape and replace each hook with your database's equivalent. Save it as my_store.py.

import json
from dataclasses import replace
 
from sqlalchemy import JSON, Boolean, and_, case, false, func, literal, select, text, type_coerce
from sqlalchemy.dialects.sqlite import insert as sqlite_insert
 
from dendrux.store import AppendArray, InsertPlan, SelectSource, SetFields, SQLStateStore
 
 
class MiniStore(SQLStateStore):
    dialect_names = ("sqlite",)  # engine.dialect.name values this store accepts
 
    supports_update_returning = True
    supports_insert_returning = True
    supports_row_locks = False  # SQLite has one writer and no SELECT ... FOR UPDATE
    ddl_transactional = True
 
    def __init__(self, engine, *, live_updates=None):
        super().__init__(engine, live_updates=live_updates)  # optional: see "Live updates"
        # Say how many processes may share the data.
        self.capabilities = replace(self.capabilities, deployment="single_host")
 
    # -- writes -----------------------------------------------------------------------------
 
    def insert_if_absent_stmt(self, table, source, *, conflict, returning=()):
        stmt = sqlite_insert(table)
        if isinstance(source, SelectSource):
            stmt = stmt.from_select(source.columns, source.select)
        elif source:  # an empty source means the base supplies the values when it runs this
            stmt = stmt.values(source)
        stmt = stmt.on_conflict_do_nothing(
            index_elements=conflict.target, index_where=conflict.predicate
        )
        if returning:
            stmt = stmt.returning(*returning)
        return InsertPlan(stmt, "targeted_ignore")
 
    def upsert_setting(self, table, *, key, value, updated_at):
        return (
            sqlite_insert(table)
            .values(key=key, value=value, updated_at=updated_at)
            .on_conflict_do_update(
                index_elements=[table.c.key], set_={"value": value, "updated_at": updated_at}
            )
        )
 
    # -- JSON ------------------------------------------------------------------------------
 
    def _path(self, path):
        result = "$"
        for part in path:
            if isinstance(part, int):
                result += f"[{part}]" if part >= 0 else f"[#{part}]"
            else:
                result += "." + json.dumps(part, ensure_ascii=False)
        return result
 
    def json_get(self, col, path, as_type):
        target = self._path(path)
        kind = func.json_type(col, target)
        value = func.json_extract(col, target)
        if as_type == "json":
            encoded = case(
                (kind == "true", "true"), (kind == "false", "false"), else_=func.json_quote(value)
            )
            return type_coerce(encoded, JSON)
        if as_type == "str":
            return case((kind == "text", value)).collate("BINARY")
        if as_type == "int":
            return case((kind == "integer", value))
        if as_type == "bool":
            return type_coerce(case((kind == "true", True), (kind == "false", False)), Boolean)
        raise ValueError("unsupported JSON read type")
 
    def json_contains(self, col, path, value):
        target = self._path(path)
        if isinstance(value, (dict, list)):
            # Structural containment: every member of `value` appears, with the same JSON type.
            wanted = func.json_tree(json.dumps(value)).table_valued("fullkey", "value", "type")
            actual = func.json_tree(func.json_extract(col, target)).table_valued(
                "fullkey", "value", "type"
            )
            missing = (
                select(literal(1))
                .select_from(wanted)
                .where(
                    ~select(literal(1))
                    .select_from(actual)
                    .where(
                        actual.c.fullkey == wanted.c.fullkey,
                        actual.c.type == wanted.c.type,
                        actual.c.value.is_not_distinct_from(wanted.c.value),
                    )
                    .exists()
                )
                .exists()
            )
            return func.coalesce(and_(func.json_type(col, target).is_not(None), ~missing), false())
        kind = (
            "null" if value is None
            else "true" if value is True
            else "false" if value is False
            else "text" if isinstance(value, str)
            else "integer" if isinstance(value, int)
            else "real"
        )
        return func.coalesce(
            and_(
                func.json_type(col, target) == kind,
                func.json_extract(col, target).is_not_distinct_from(value),
            ),
            false(),
        )
 
    def json_is_null(self, col, path):
        return func.coalesce(func.json_type(col, self._path(path)) == "null", false())
 
    def json_patch_expr(self, col, patch):
        base = func.coalesce(func.nullif(col, "null"), "{}")
        if isinstance(patch, SetFields):
            args = [
                item
                for key, value in patch.fields.items()
                for item in (self._path((key,)), func.json(json.dumps(value)))
            ]
            return func.json_set(base, *args) if args else base
        if isinstance(patch, AppendArray):
            return func.json_insert(
                base, self._path(patch.path) + "[#]", func.json(json.dumps(patch.value))
            )
        return None  # PatchLoopState: the base reads, modifies and writes under a lock
 
    # -- locks ------------------------------------------------------------------------------
 
    async def serialize_in_tx(self, session, *, owner):
        # One writer at a time already serialises note counting here. A database with row
        # locks can leave this to the default, which locks the run row.
        return None
 
    async def reserve_parent_writer(self, session, run_id):
        # Without row locks, take the write lock on the run's row before checking it.
        await session.execute(
            text("UPDATE agent_runs SET cancel_requested = cancel_requested WHERE id = :run"),
            {"run": run_id},
        )

Every hook not shown keeps the base's default, which works for most databases: lock_clause, id_in, child_page_query, classify_db_error, connection_checks, blob_now, metadata_filter_predicate, pause_projection, json_conversion_error, event_sequence_expr, inserted_event_sequence and idempotency_scope_predicate. The hook reference says what each does and when to override it.

Plug it in

Build the store from an engine you create once and own, then pass it wherever a store goes. There is no registry and no other call:

from dendrux import Agent
from dendrux.store import RunStore
 
store = MiniStore(engine)
agent = Agent(provider=..., prompt=..., state_store=store)
reads = RunStore(store)  # read primitives: get_pending, follow, get_event_view, and the rest

Agent(state_store=...) and RunStore(store) accept any store. store_for_engine(engine), RunStore.from_engine(engine) and Agent(database_url=...) pick one of the three built-in classes from the engine's dialect and know nothing about yours, so use the two lines above and keep the store object around. RunStore.from_engine also never listens for live updates; wrap your own store with RunStore(store). The application owns the engine and disposes it.

Create the schema

Create the tables once, at deploy time, from the models. A fresh schema made this way starts at storage format 2 (the full 0.6 behaviour), so no activation step is needed:

from dendrux.db.models import Base
 
 
async def create_schema(engine) -> None:
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)

Call it once with asyncio.run(create_schema(engine)), or await it inside your setup code.

dendrux db migrate, check, status and activate run Alembic migrations that Dendrux writes and tests for PostgreSQL, SQLite and MySQL/MariaDB only. For your own dialect you own the schema: create it with the snippet above, and when a later Dendrux release changes the models, apply the matching change to your database yourself. The kit builds its test schema the same way (see Provide the fixtures); migrations are not part of certification.

Feature flags

The flags are independent. Set each one from what your database really does, and let the kit tell you if you were wrong. connection_checks may override a flag at first use (the MySQL store resolves supports_insert_returning that way, because MariaDB has it and MySQL does not).

Hook reference

Every hook below is a method of SQLStateStore that you may override. A hook never commits, retries or validates: the base owns the transaction. Types are the names from dendrux.store; Table, ColumnElement, Select, Executable and Result come from SQLAlchemy.

Statements and expressions

Hooks that run a statement

These are async and receive the session of the transaction the base opened. They must not commit.

Reading errors

Optional

Two rules hold for every hook:

• Tenant and identifier equality must be exact. "t" and "t " are different tenants, and so are "T" and "t". If your database pads or folds case when it compares text, choose a binary, no-pad collation for every identifier, tenant, key and idempotency column. The kit checks this.
• Fail closed. If a hook cannot decide, it refuses. A hook the base requires and you did not implement raises NotImplementedError naming the hook.

Declare what you promise

A store states what it supports in capabilities:

from dataclasses import replace
 
class MyStore(SQLStateStore):
    def __init__(self, engine):
        super().__init__(engine)
        self.capabilities = replace(self.capabilities, deployment="multi_server")

The base class declares everything it implements. You set deployment. A store without a deployment cannot be certified. Live updates are not a capability: they are an optional argument of the store, covered in the next section.

Live updates in a custom store

Live updates let other processes learn that a run changed, so their follow() wakes at once instead of at the next poll. They are optional: a store without them keeps working, and followers poll. The hints are lossy and the database stays the truth; see Live updates for what they do for a deployment and for the transports (PostgresUpdates, RedisUpdates, your own).

A SQLStateStore subclass only has to pass the keyword through, as MiniStore above does:

class MyStore(SQLStateStore):
    def __init__(self, engine, *, live_updates=None):
        super().__init__(engine, live_updates=live_updates)

The base class binds the instance, marks the runs each write touches, hands their ids to it once the transaction's COMMIT was issued, flushes when the outermost store call returns, and closes it in await store.close() (which never disposes your engine). live_updates=None or False turns it off. live_updates=True is the store's own transport, and a subclass has none, so it raises ValueError; pass an instance such as RedisUpdates("redis://..."). A LiveUpdates instance belongs to one store.

A store that is not built on SQLStateStore (any object that implements the StateStore protocol) can announce too. It must:

1. Accept live_updates in its constructor, call live_updates.bind(self) and expose the instance as a live_updates attribute. RunStore(store) reads that attribute to decide whether to listen.
2. Call live_updates.enqueue(run_ids) for every write that changes what a follower, a status read or a run list shows, once that write's commit was issued, even if the commit then fails: the write may have landed. Do not enqueue for a write that failed before its commit or was proved not to have applied.
3. Call await live_updates.flush() at the end of each such public call, on the success path only. A call that raises should keep its original exception and leave its ids pending for the next successful call or for close().
4. Call await live_updates.close() from its own close().

enqueue takes an iterable of run-id strings and never blocks. It drops ids that are empty, too long or beyond its pending bound, and counts them, because a lost hint only costs a poll. A bare str is the one input it refuses: enqueue("01J...") raises TypeError, because an id passed by mistake would otherwise be split into single characters. flush() never raises except for the caller's own cancellation, waits at most flush_deadline_s, and does nothing while the transport is backing off after a failure.

Followers read a store through two optional batched methods and fall back to one read per run, under the same query budget, when the store has none:

Both take run ids that the caller has already authorized. They are reader building blocks, not authorization APIs, and RunStore does not expose them. SQLStateStore implements both.

Certify it. Add the optional dendrux_live_store_factory fixture (see Provide the fixtures). The kit then gives your store a recording LiveUpdates and checks that a committed write is announced only after a second connection can see it, that a write proved not to have applied announces nothing, and that a write whose COMMIT was issued is announced at the next flush even when the call failed or was cancelled. Without the fixture the report says "live updates: not tested" and the verdict carries that caveat.

Provide the fixtures

The kit runs against fixtures you write in your own conftest.py. They return plain callables, so they work with any async test setup.

Here is the conftest.py that goes with the store above. One small object keeps every engine the fixtures create, so they are all disposed at the end, and a second instance reuses the engine's URL object (a URL turned into a string hides the password). Base.metadata.create_all is all the schema setup the kit needs:

import pytest
import pytest_asyncio
from sqlalchemy.ext.asyncio import create_async_engine
 
from dendrux.db.models import Base
from dendrux.testing.store_kit.sql import SqlClock, SqlFaultInjector
 
from my_store import MiniStore
 
 
class World:
    def __init__(self, directory):
        self.directory = directory
        self.engines = []
 
    def engine(self, url):
        engine = create_async_engine(url, connect_args={"check_same_thread": False})
        self.engines.append(engine)
        return engine
 
    async def fresh(self, live_updates=None):
        url = f"sqlite+aiosqlite:///{self.directory / f'kit{len(self.engines)}.db'}"
        engine = self.engine(url)
        async with engine.begin() as conn:
            await conn.run_sync(Base.metadata.create_all)  # tables, and a schema at format 2
        return MiniStore(engine, live_updates=live_updates)
 
    async def another(self, store):
        # store._engine.url is a URL object: it keeps the password. Do not use str(url).
        return MiniStore(self.engine(store._engine.url))
 
 
@pytest_asyncio.fixture
async def world(tmp_path):
    world = World(tmp_path)
    yield world
    for engine in world.engines:
        await engine.dispose()
 
 
@pytest.fixture
def dendrux_store_factory(world):
    return world.fresh
 
 
@pytest.fixture
def dendrux_live_store_factory(world):
    return world.fresh  # the kit passes the LiveUpdates the store must announce through
 
 
@pytest.fixture
def dendrux_store_reopen(world):
    return world.another  # a restarted process: a new instance over the same data
 
 
@pytest.fixture
def dendrux_store_second_instance(world):
    return world.another  # a second live instance has its own engine over the same data
 
 
@pytest.fixture
def dendrux_clock(world):
    return SqlClock(lambda: world.engines[0])  # ages every stored run clock
 
 
@pytest.fixture
def dendrux_fault_injector():
    injector = SqlFaultInjector()
    yield injector
    injector.close()

SqlClock and SqlFaultInjector work on any SQLAlchemy engine, so a SQLStateStore subclass gets both without writing them. If your clock instead moves the store's time source forward (so stored timestamps stay as they are), give it a now() method returning the store's current time: the kit computes lease cutoffs from it. A clock that ages stored timestamps needs no now().

A fault injector has arm(store, point, kind=, after=, sticky=), fired(store) and disarm(store). fired must count the cuts that really happened: the kit asserts that every cut it asked for fired, so an injector that does nothing fails certification. To inject lost connections the kit asks the injector which forms of disconnect exist for your store, through an optional disconnect_forms(store) method (the default is ("invalidated",)). An injector that returns an empty list leaves the claim contracts untested, and the store is NOT CERTIFIED for what they guard.

The tests create their schema from the models, so your factory must too. Make sure the schema starts at format 2 (a fresh create_all does). A factory whose runs come out at format 1 while the store declares the four format-2 capabilities cannot be certified: every format-2 contract would skip, and the kit says so rather than passing. Schema migrations are tested separately.

Run the kit

pip install "dendrux[testkit]"
pytest -p dendrux.testing.store_kit

-p dendrux.testing.store_kit loads the plugin and collects the contracts; you do not copy any test files. Installing dendrux registers a tiny bootstrap module with pytest. It does not import dendrux or load the kit until you switch it on, so other pytest sessions in the same environment are unaffected and pay no import cost. Switching it on needs pytest-asyncio (pip install "dendrux[testkit]"); without it pytest stops with a message naming the extra. Two other ways to switch it on: --dendrux-store-kit on the command line, or dendrux_store_kit = true under [tool.pytest.ini_options]. Merely having the plugin installed or listed in pytest_plugins collects nothing.

If you use pytest-asyncio in strict mode, add asyncio_default_fixture_loop_scope = "function" to your pytest settings. The kit marks its own async tests, so asyncio_mode can stay as it is.

To certify several databases, run pytest once per database (one CI job each) and let your fixtures read the database URL from the environment.

The exit code is the verdict. A session that prints NOT CERTIFIED, including an incomplete run, exits non-zero, so CI can gate on it. A session that prints CERTIFIED with caveats or CERTIFIED exits zero (unless a test failed, which exits non-zero as usual). If no dendrux_store_factory fixture exists the kit fails one test with a message that says what to define, then stops, instead of failing every contract the same way.

How certification works

The kit reads the capabilities and the deployment your store declares:

Skips the store causes. Some contracts skip because of something the store controls. The kit lists them in the summary, per capability, under "skipped by the store", and any declared capability they guard is NOT CERTIFIED:

• The store declares fenced_writes, retry_ids, checkpointing and atomic_completion but creates storage-format-1 runs, so every format-2 contract skips. Fix the factory's schema (use create_all) or declare less.
• The fault injector offers no disconnect form (disconnect_forms returns an empty list), so lost commit acknowledgements cannot be injected into claims.
• The store has no release_run_claim, so releasing a claim after a lost acknowledgement is not tested.

Skips that are legitimate stay ordinary: a capability you did not declare, a disconnect form that does not exist for your dialect (as long as another form ran), and a second instance on a single_process store.

The basic contracts run on whatever storage format your store creates: format 1 when the four format-2 flags (fenced_writes, retry_ids, checkpointing, atomic_completion) are not all declared, format 2 otherwise.

If two different store classes share a class name (two packages each with a PostgresStore), the summary keeps them apart and prints the module path.

At the end of the session the kit prints a summary per store. This is what a store that passes everything the protocol can observe looks like (the unverified save_usage line is expected, and it makes the verdict a caveated one):

=================== dendrux store kit: certification summary ===================
MiniStore (deployment=single_host, capabilities=fenced_writes,retry_ids,checkpointing,atomic_completion,recovery)
  declaration        3 passed
  basic              34 passed
  agent_flows        9 passed
  fenced_writes      22 passed
  retry_ids          13 passed, 1 skipped
  checkpointing      12 passed
  atomic_completion  11 passed
  recovery           13 passed
  concurrency        16 passed
  crash_recovery     54 passed, 16 skipped
  live_updates       7 passed
  unverified: save_usage retry identity: usage rows cannot be read through the protocol, so a duplicate row is not detected
  CERTIFIED with caveats: declaration, basic, agent_flows, fenced_writes, retry_ids, checkpointing, atomic_completion, recovery, concurrency, crash_recovery, live_updates (caveats: unverified: save_usage retry identity: usage rows cannot be read through the protocol, so a duplicate row is not detected)

The numbers change as contracts are added; the shape does not. The skipped crash_recovery contracts are the disconnect forms SQLite does not have, and the skipped retry_ids contract is the one for a save_usage identity the protocol cannot read back.

What each group checks

When a contract fails

• A fenced writer accepted a stale epoch. Your update does not include the epoch in its WHERE clause, or it is written outside the base's fenced helpers.
• Retry stored twice. The identity is not part of the unique constraint, or insert_if_absent_stmt ignores clashes on the wrong key.
• Two winners for a claim. The conditional update is not atomic on your database, or supports_update_returning is True when it should be False.
• Gaps or duplicates in event sequences. insert_if_absent_stmt or serialize_in_tx does not serialize writers of one run.
• The note cap is exceeded. serialize_in_tx does not hold across the count and the insert.
• Tenants "t" and "t " collide. Your column collation pads or folds case.
• An atomic operation left half its writes. A write happened outside the single transaction the base opens. Do not commit inside a hook.

Ask for help

Open an issue with the kit's summary and the failing contract names. The summary contains no run data or driver text.