PostgresSaver

Create a new PostgresSaver instance from a connection string.

Set up the checkpoint database asynchronously.

This method creates the necessary tables in the Postgres database if they don't already exist and runs database migrations. It MUST be called directly by the user the first time checkpointer is used.

List checkpoints from the database.

This method retrieves a list of checkpoint tuples from the Postgres database based on the provided config. The checkpoints are ordered by checkpoint ID in descending order (newest first).

Get a checkpoint tuple from the database.

This method retrieves a checkpoint tuple from the Postgres database based on the provided config. If the config contains a checkpoint_id key, the checkpoint with the matching thread ID and timestamp is retrieved. Otherwise, the latest checkpoint for the given thread ID is retrieved.

Save a checkpoint to the database.

This method saves a checkpoint to the Postgres database. The checkpoint is associated with the provided config and its parent config (if any).

Store intermediate writes linked to a checkpoint.

This method saves intermediate writes associated with a checkpoint to the Postgres database.

Delete all checkpoints and writes associated with a thread ID.

Prune checkpoints for the given threads.

"delete" removes every checkpoint, write and blob of the threads, in one transaction.

"keep_latest" keeps the latest checkpoint of each namespace (the one get_tuple loads, plus its parent when it predates v4 and has pending sends), with its pending writes and the blobs it references, and deletes the rest. A kept checkpoint whose parent it deletes is left with no parent. A subgraph call's namespace is deleted with the root checkpoint the call started from, except a checkpointer=True subgraph's, which keeps its latest checkpoint. Each thread is pruned in its own transaction, on one snapshot of it, so a run appending to the thread meanwhile is safe; forking or replaying from a checkpoint being deleted is not. A thread that fails because another transaction changed it leaves the threads before it pruned, and retrying is safe.

"keep_latest" refuses threads that use a DeltaChannel, since it rebuilds the latest value from older checkpoints: if any requested thread does, nothing is pruned. A thread that starts using one before its turn in the batch is refused then, after the threads before it.

Fast-path override of BaseCheckpointSaver.get_delta_channel_history.

Two-stage query, both stages cover ALL requested channels:

• Stage 1 (paged): dynamic SELECT over checkpoints with three columns per channel: its version, an EXISTS probe for a stored blob at that version, and its inline value. Pages newest-first by checkpoint_id, starting at the target; page size is _DELTA_PAGE_SIZE. Stops paging when every channel has found its seed or a page comes back short.

• Stage 2 (per-channel UNION ALL): one branch per channel reading checkpoint_writes filtered to that channel's specific chain_cids, plus one branch per channel that has a seed reading checkpoint_blobs for that channel + version. Avoids the over-fetch of a single channel = ANY(channels) filter when channels have different chain depths.