Checkpointers

When to use this

A checkpointer persists graph state between requests so conversations survive server restarts, threads can be paused and resumed, and multiple turns stay coherent. Without a checkpointer the graph uses an in-memory default that is reset every call.

Import paths

from tenxgraph.storage.checkpointer import BaseCheckpointer, InMemoryCheckpointer
# Optional — requires asyncpg
from tenxgraph.storage.checkpointer import PgCheckpointer
# Optional — requires aiosqlite
from tenxgraph.storage.checkpointer import SqliteCheckpointer

BaseCheckpointer[StateT]

Abstract base class for all checkpointer implementations. Provides both async and sync method pairs.

Abstract async methods

Each abstract method must be implemented by a subclass:

Sync wrappers

For every async axxx() method there is a sync xxx() wrapper that calls asyncio.run(). Use these only from non-async contexts (e.g. a management script):

checkpointer.put_state(config, state)
checkpointer.get_state(config)

Wiring into a graph

from tenxgraph.storage.checkpointer import InMemoryCheckpointer

app = graph.compile(checkpointer=InMemoryCheckpointer())

InMemoryCheckpointer

In-process dictionary-based storage. Zero dependencies.

from tenxgraph.storage.checkpointer import InMemoryCheckpointer

checkpointer = InMemoryCheckpointer()
app = graph.compile(checkpointer=checkpointer)

When to use:

• Unit tests and CI.
• Local development when you don’t need state to survive a restart.
• Ephemeral single-process jobs.

When NOT to use:

• Any multi-process or multi-worker deployment.
• Production applications where conversation history must survive crashes.

Storage behaviour

All access is guarded by per-bucket asyncio.Lock instances for safe concurrent use within a single process.

PgCheckpointer

PostgreSQL-backed checkpointer with optional Redis caching. Production-grade.

from tenxgraph.storage.checkpointer import PgCheckpointer

checkpointer = PgCheckpointer(
    postgres_dsn="postgresql://user:pass@localhost:5432/mydb",
    redis_url="redis://localhost:6379/0",   # optional
)

await checkpointer.asetup()   # creates tables if they don't exist
app = graph.compile(checkpointer=checkpointer)

Constructor parameters

ID types

PgCheckpointer adapts its schema based on the id_type registered by the compiled graph’s id_generator:

This is set automatically. You do not set it directly.

Schema migration

asetup() creates the required tables if they do not exist. It is idempotent — safe to call on every startup.

SqliteCheckpointer

Single-file SQLite checkpointer. Stores everything — durable state, the hot “realtime” state cache, generic TTL cache, messages, and threads — in one local .db file. No Postgres, no Redis. Fully async via aiosqlite.

from tenxgraph.storage.checkpointer import SqliteCheckpointer

# Defaults to ~/.10xgraph/checkpointer.db; pass a path to override.
checkpointer = SqliteCheckpointer("agent_state.db")

await checkpointer.asetup()   # creates tables (also runs lazily on first use)
app = graph.compile(checkpointer=checkpointer)

Constructor parameters

When to use:

• Client-side / embedded agents. A desktop app that ships a Python sidecar (Tauri, Electron, PyInstaller) or a local CLI agent — the checkpointer runs entirely on the user’s machine, right next to the app.
• Dedicated-room deployments. Each user (or tenant, or session) has their own process and their own .db file, so there is exactly one writer per database.

When NOT to use:

• Multi-user servers where many users share one backend. SQLite serializes writers and does not scale horizontally. Reach for PgCheckpointer (Postgres + optional Redis) instead.

Design notes

• One file, no external services. The realtime state cache lives in a separate SQLite table (af_state_cache) rather than Redis; durable state lives in af_states. Reads check the cache table first, then fall back to durable state — the same two-layer read path as PgCheckpointer, collapsed into one file.
• Custom state is preserved. State is serialized with an embedded __class_path__ and reconstructed into its exact AgentState subclass on read, matching PgCheckpointer.
• Concurrency. A single persistent connection is opened lazily with PRAGMA journal_mode=WAL; writes are serialized behind an asyncio.Lock. Call await checkpointer.arelease() (or checkpointer.release()) at shutdown to close it.
• Isolation. Data is keyed purely by thread_id; user_id is not required. alist_threads returns every thread in the file, which matches the one-database-per-user model.

Writing a custom checkpointer

from typing import Any
from tenxgraph.storage.checkpointer import BaseCheckpointer
from tenxgraph.core.state import AgentState, Message
from tenxgraph.utils.thread_info import ThreadInfo

class DynamoDBCheckpointer(BaseCheckpointer):

    async def asetup(self) -> Any:
        # Create DynamoDB tables
        ...

    async def aput_state(self, config: dict, state: AgentState) -> AgentState:
        thread_id = config["thread_id"]
        # Serialise and PUT to DynamoDB
        ...
        return state

    async def aget_state(self, config: dict) -> AgentState | None:
        thread_id = config["thread_id"]
        # GET from DynamoDB and deserialise
        ...

    async def aclear_state(self, config: dict) -> Any:
        ...

    async def aput_state_cache(self, config: dict, state: AgentState) -> Any:
        ...  # in-memory dict, no external call

    async def aget_state_cache(self, config: dict) -> AgentState | None:
        ...

    async def aput_messages(
        self, config: dict, messages: list[Message], metadata: dict | None = None
    ) -> Any:
        ...

    async def aget_message(self, config: dict, message_id: str | int) -> Message:
        ...

    async def alist_messages(
        self,
        config: dict,
        search: str | None = None,
        offset: int | None = None,
        limit: int | None = None,
    ) -> list[Message]:
        ...

    async def adelete_message(self, config: dict, message_id: str | int) -> Any | None:
        ...

    async def aput_thread(self, config: dict, thread_info: ThreadInfo) -> Any | None:
        ...

    async def aget_thread(self, config: dict) -> ThreadInfo | None:
        ...

    async def alist_threads(
        self,
        config: dict,
        search: str | None = None,
        offset: int | None = None,
        limit: int | None = None,
    ) -> list[ThreadInfo]:
        ...

    async def aclean_thread(self, config: dict) -> Any | None:
        ...

    async def arelease(self) -> Any | None:
        ...  # close connections

Config dictionary

All checkpointer methods accept a config dict. The required key is:

config = {"thread_id": "session-abc123"}

Additional keys used internally:

• user_id — scopes message searches by user.
• run_id — tracks a specific invocation.

Common errors