How PgCheckpointer works

PgCheckpointer implements a two-layer storage model: a durable layer in PostgreSQL with full versioned history, and a hot cache layer in Redis with TTL-controlled entries. Understanding how state flows through these layers is essential for production deployment and debugging concurrency issues.

The two-layer model

PostgreSQL is the source of truth. Redis is expendable: if it crashes or the cache entry expires, the next read simply falls back to Postgres. Conversely, losing the Redis cache does not lose data.

The write path: durable first, then cache

When the graph loop checkpoints state, it calls PgCheckpointer.aput_checkpoint() and then aput_state_cache(), two sequential operations:

1. Durable write to PostgreSQL

State and messages are written in a single database transaction that:

1. Creates the thread row if absent and locks it (SELECT ... FOR UPDATE) to serialize writers, so concurrent workers on the same thread cannot collide. When isolation is active, it also rejects a thread owned by a different user.
2. Performs an optimistic version check: reads the highest version in the states table and, if the run recorded a version when it read the state, raises StaleStateError if the two differ because another run committed in the meantime
3. If the check passes, inserts a new row with version = current_version + 1 and the serialized state data
4. Inserts any new messages into the messages table
5. Prunes states older than the state_history_limit window (default 20 rows)
6. Commits the transaction

If PostgreSQL fails before commit, the transaction is rolled back and no state is persisted. The run’s thread is left unchanged.

2. Cache refresh to Redis

After PostgreSQL commits, the graph loop writes the same state to Redis under the key state_cache:{thread_id}:{user_id} with the configured TTL. This cache write is atomic via a Lua script with a version guard: the script only writes if the incoming version is equal to or newer than the cached one, so a stale run cannot overwrite a newer cached state.

If the Redis write fails, the failure is logged and ignored. The durable state is safe in PostgreSQL, and the next read will fall back to Postgres and refresh the cache.

Write sequence guarantees

The PostgreSQL-first ordering ensures the cache never holds state that is not durable. If a crash occurs between the Postgres commit and the Redis write, the Redis cache remains stale (or empty), but the durable state is safe. The next read will detect the cache miss and fetch from Postgres.

The read path: cache-aware with atomic fallback

At the start of a run, the graph calls PgCheckpointer.aget_state_cache() (falling back to aget_state()) to retrieve the thread’s latest state:

1. Redis hit: Read state_cache:{thread_id}:{user_id} from Redis. Extract the embedded version number (stored as __checkpoint_version__ in the cached JSON). Restore it into config['_checkpoint_version'] so a subsequent write can use it for the version check. Return the deserialized state.

2. Redis miss or error: Execute a Postgres read:

   • Query SELECT version, state_data FROM states WHERE thread_id = ? ORDER BY version DESC, state_id DESC LIMIT 1, scoped to the owning user through the threads table when isolation is active
   • Deserialize the state and extract the version
   • Populate config['_checkpoint_version'] with this version
   • Return the state

3. Thread not found in Postgres: Return None. The graph will initialize a fresh AgentState and use the incoming messages (if any) to populate it.

4. Cache refreshed on miss: If the Postgres read succeeded, immediately write the result back to the Redis cache (with TTL) so the next call to this thread hits the cache.

This two-stage read minimizes Postgres load in the common case (cache hits) while guaranteeing correctness (the version from either cache or Postgres is used for the next write’s compare-and-swap).

Versions and optimistic concurrency control

Every call to aput_checkpoint() or aput_state() performs a compare-and-swap:

1. The run reads a thread’s state at the start and records its version in config['_checkpoint_version']
2. The run executes its graph
3. At each checkpoint, the write checks if that version still matches the thread’s highest version in Postgres
4. If another run committed a new version in the meantime, the write raises StaleStateError instead of updating

The check runs inside the transaction, after the thread row is locked and before the insert. If the check fails:

• The transaction is rolled back (no state is persisted)
• The cached entry is invalidated (dropped from Redis) to prevent the thread from staying stuck on stale data
• StaleStateError is raised to the caller

This prevents lost updates: each run’s writes are serialized by version number, and concurrent runs that started from the same version cannot both commit.

Example: three concurrent runs

Run A          Run B          Run C
├─ Read @ v=0
├─ Read @ v=0
│  ├─ Read @ v=0
│  │  ├─ Execute (A)
│  │  ├─ Write → v=1 ✓
│  │  └─ Cache v=1
│  ├─ Execute (B)
│  ├─ Write expected v=0, found v=1 → StaleStateError
│  └─ Cache invalidated
├─ Execute (C)
└─ Write expected v=0, found v=1 → StaleStateError
   └─ Cache invalidated

Run A succeeds. Runs B and C both fail because they started from v=0 but A already advanced to v=1. Their cache entries are dropped so they do not poison the next read. The calling code can retry them if needed.

History pruning and bounded growth

By default, PgCheckpointer keeps 20 state snapshots per thread (state_history_limit=20). After each successful write, older rows are deleted:

# Pseudocode from _write_state_row()
new_version = current_version + 1
INSERT INTO states (thread_id, version, state_data, meta) VALUES (...)
DELETE FROM states
  WHERE thread_id = ?
  AND version <= (new_version - state_history_limit)

This prevents the states table from growing indefinitely. The 20-row window is enough for debugging the recent execution history but not an audit log. If you need to preserve all history, export it before it is pruned, or increase state_history_limit (at the cost of more Postgres storage).

Messages are stored separately in the messages table and are not automatically pruned by version limits (only by explicit deletion or thread cleanup).

Database schema and table structure

The PgCheckpointer.asetup() method creates these tables (in the specified schema, default public):

The states table has a unique index on (thread_id, version) to enforce one row per version per thread. Writes lock the thread’s row in threads to serialize concurrent writers.

User isolation and scoping

By default, enforce_user_isolation=True. The states and messages tables carry no user_id; ownership lives on threads, so queries are scoped through it (thread_id IN (SELECT thread_id FROM threads WHERE user_id = ?)) using the user_id in the config dict. This prevents one user from reading or writing another user’s thread, even if they know the thread_id. When the API server supplies an authorization policy in the config, that policy decides whether isolation applies.

PgCheckpointer raises a ValueError on a read or write if user_id is missing from the config. The API server fills user_id from the authenticated request (or sets it to "anonymous" if no auth is configured). Turn off isolation only for single-tenant deployments where all threads are logically shared.

Cache invalidation on conflict

When a write fails the optimistic version check, PgCheckpointer immediately calls _invalidate_state_cache() to drop the cached entry:

await self.redis.delete(self._get_thread_key(thread_id, user_id))

This prevents a poisoned cache: if Run A loses its race and reads the cached state that Run B wrote (which was built on a newer version), Run A’s next write will again fail because it is trying to commit the older version. By dropping the cache, Run A’s next read will fetch from Postgres and get the latest version, allowing it to retry successfully.

Storage configuration patterns

Development: InMemoryCheckpointer

For local development and unit tests, use InMemoryCheckpointer. State lives in process memory; no Postgres or Redis required:

from tenxgraph.storage.checkpointer import InMemoryCheckpointer
checkpointer = InMemoryCheckpointer()

Single-user: SqliteCheckpointer

For client-side agents (desktop apps, CLIs, embedded systems), use SqliteCheckpointer. Durable state and messages live in a single SQLite file:

from tenxgraph.storage.checkpointer import SqliteCheckpointer
checkpointer = SqliteCheckpointer("./agent_state.db")

Production: PgCheckpointer

For multi-worker services, use PgCheckpointer with Postgres and Redis. See Set up checkpointing for detailed configuration.

Related concepts

• Durability and concurrency: For handling StaleStateError, thread isolation, and optimistic concurrency patterns, see Durability and concurrency.
• Thread lifecycle: For the config dict, thread_id, and user_id, see Checkpointing and threads.
• Long-term memory: The checkpointer is thread-scoped. For memories shared across threads, see Long-term memory.
• Replay-safe tools: How the tool ledger uses durable writes to prevent tools from running twice on resume, see Replay-safe tools.

Related pages