Checkpointing API

Conversation checkpointing lets an agent snapshot its state and rewind to an earlier turn. Enable it via include_checkpoints=True on create_deep_agent. See Checkpointing for the conceptual overview.

Checkpoint

pydantic_deep.toolsets.checkpointing.Checkpoint dataclass

Immutable snapshot of conversation state at a point in time.

Attributes:

Name	Type	Description
id	str	Unique identifier (uuid4).
label	str	Human-readable label (e.g. "auto-3", "before-refactor").
turn	int	Model-request counter when the checkpoint was saved.
messages	list[ModelMessage]	Shallow copy of the message history (ModelMessage is immutable).
message_count	int	Number of messages in the snapshot.
created_at	datetime	When the checkpoint was created.
metadata	dict[str, Any]	Optional metadata (e.g. {"last_tool": "write_file"}).

CheckpointStore

pydantic_deep.toolsets.checkpointing.CheckpointStore

Bases: Protocol

Protocol for checkpoint storage backends.

save(checkpoint) async

Save or overwrite a checkpoint.

get(checkpoint_id) async

Get a checkpoint by ID.

get_by_label(label) async

Get a checkpoint by label (returns first match).

list_all() async

List all checkpoints ordered by creation time.

remove(checkpoint_id) async

Remove a checkpoint by ID. Returns True if it existed.

remove_oldest() async

Remove the oldest checkpoint. Returns True if one was removed.

count() async

Return the number of stored checkpoints.

clear() async

Remove all checkpoints.

InMemoryCheckpointStore

pydantic_deep.toolsets.checkpointing.InMemoryCheckpointStore

In-memory checkpoint store (default).

Checkpoints are stored in a dict keyed by ID. Iteration order matches insertion order (Python 3.7+), so remove_oldest() pops the first key.

save(checkpoint) async

Save or overwrite a checkpoint.

get(checkpoint_id) async

Get a checkpoint by ID.

get_by_label(label) async

Get a checkpoint by label (returns first match).

list_all() async

List all checkpoints ordered by creation time.

remove(checkpoint_id) async

Remove a checkpoint by ID.

remove_oldest() async

Remove the oldest checkpoint by created_at.

Ordered by created_at (not insertion order) so this store and :class:FileCheckpointStore evict the same checkpoint - keeping the two interchangeable when clock skew or relabeling reorders entries.

count() async

Return the number of stored checkpoints.

clear() async

Remove all checkpoints.

FileCheckpointStore

pydantic_deep.toolsets.checkpointing.FileCheckpointStore

Persistent checkpoint store using JSON files.

Each checkpoint is stored as {directory}/{id}.json. Messages are serialized using pydantic-ai's ModelMessagesTypeAdapter.

Parameters:

Name	Type	Description	Default
directory	str | Path	Directory to store checkpoint files.	required

save(checkpoint) async

Save checkpoint to a JSON file.

get(checkpoint_id) async

Load checkpoint from JSON file.

get_by_label(label) async

Find a checkpoint by label (scans all files).

list_all() async

List all checkpoints ordered by creation time.

remove(checkpoint_id) async

Remove a checkpoint file.

remove_oldest() async

Remove the oldest checkpoint file.

count() async

Count checkpoint files.

clear() async

Remove all checkpoint files.

CheckpointMiddleware

pydantic_deep.toolsets.checkpointing.CheckpointMiddleware dataclass

Bases: AbstractCapability[Any]

Capability that auto-saves conversation checkpoints.

Uses before_model_request (for every_turn frequency) and after_tool_execute (for every_tool frequency) hooks to automatically snapshot the conversation.

Per-run state isolation via for_run() ensures concurrent agent runs don't share turn counters.

Parameters:

Name	Type	Description	Default
store	CheckpointStore | None	Fallback checkpoint store (used if deps has no store).	None
frequency	str	When to auto-checkpoint: "every_turn" - before each model request, "every_tool" - after each tool call, "manual_only" - no auto-checkpoints.	'every_tool'
max_checkpoints	int	Maximum number of checkpoints to keep.	20

for_run(ctx) async

Return a fresh instance with isolated per-run state.

before_model_request(ctx, request_context) async

Track messages and optionally auto-checkpoint before model calls.

after_tool_execute(ctx, *, call, tool_def, args, result) async

Auto-checkpoint after tool calls (when frequency is every_tool).

Snapshots ctx.messages (the live history, which already contains the ModelResponse with this tool's ToolCallPart) plus a ToolReturnPart carrying the result, so rewinding to a tool-* checkpoint restores a state that actually includes the tool call and its result.

CheckpointToolset

pydantic_deep.toolsets.checkpointing.CheckpointToolset

Bases: FunctionToolset[Any]

Toolset giving the agent manual checkpoint controls.

Tools

• save_checkpoint: Label the most recent auto-checkpoint
• list_checkpoints: Show all saved checkpoints
• rewind_to: Rewind conversation to a checkpoint (raises RewindRequested)

__init__(*, store=None, id='deep-checkpoints', descriptions=None)

Initialize the checkpoint toolset.

Parameters:

Name	Type	Description	Default
store	CheckpointStore | None	Fallback checkpoint store (used if deps has no store).	None
id	str	Toolset identifier.	'deep-checkpoints'
descriptions	dict[str, str] | None	Optional mapping of tool name to custom description. Supported keys: save_checkpoint, list_checkpoints, rewind_to. Any key not present falls back to the built-in description constant.	None

RewindRequested

pydantic_deep.toolsets.checkpointing.RewindRequested

Bases: Exception

Raised by the rewind_to tool to signal app-level rewind.

This exception propagates out of agent.run() / agent.iter() because pydantic-ai only catches ModelRetry and UnexpectedModelBehavior - arbitrary exceptions propagate to the caller.

The caller (e.g. the app's run loop) catches this, restores session.message_history from :attr:messages, and restarts.

Attributes:

Name	Type	Description
checkpoint_id		ID of the checkpoint to rewind to.
label		Human-readable label of the checkpoint.
messages		The message history snapshot to restore.

fork_from_checkpoint

pydantic_deep.toolsets.checkpointing.fork_from_checkpoint(store, checkpoint_id) async

Get messages from a checkpoint for forking into a new session.

This is a utility function for app-level session forking. The caller creates a new session and sets its message_history to the returned messages.

Parameters:

Name	Type	Description	Default
store	CheckpointStore	The checkpoint store to read from.	required
checkpoint_id	str	ID of the checkpoint to fork from.	required

Returns:

Type	Description
list[ModelMessage]	A copy of the checkpoint's message history.

Raises:

Type	Description
ValueError	If the checkpoint is not found.

Example

Python

messages = await fork_from_checkpoint(store, "abc123")
new_session = UserSession()
new_session.message_history = messages