Step Persistence

StepPersistence records what an agent did at each boundary, separate from whether the run can be safely resumed. It is the persistence substrate for orchestrators that delegate to sub-agents — for example, an AICA orchestrator that spawns a code_librarian to investigate one symbol, then continues that delegate’s investigation with a follow-up question.

It is not a full graph-state checkpoint. Capability-state restore, workspace snapshots, and graph-node resume are out of scope and tracked separately (see pydantic-ai-harness issues #149 and #196).

Source

The API may change between releases. Where practical, breaking changes ship with a deprecation warning.

What it gives you

Append-only step events. Every interesting boundary (run start/end, model request, tool call, failure) appends a StepEvent. A run killed mid-tool-call still leaves a usable event trail.
Continuable snapshots. A ContinuableSnapshot is saved at settled node boundaries, and a failing run saves its live at-failure history. Each snapshot carries a state: complete when every ToolCallPart has a matching result, interrupted when the capture holds unsettled tool work (e.g. a crash mid-tool-cycle). latest_snapshot and continue_run return only complete snapshots unless the caller passes include_interrupted=True. Pass the snapshot’s messages back to Agent.run(message_history=...) to continue or fork.
Tool-effect ledger. Every tool call’s lifecycle (started, completed, failed) is recorded against (run_id, tool_call_id). After a crash, a tool with a started record and no terminal update should be treated as unknown_after_crash: the side effect may or may not have happened.
Lineage metadata. conversation_id (sequence) and parent_run_id (hierarchy) are independent axes. See Three-level identity.

Quick start

import asyncio

from pydantic_ai import Agent
from pydantic_ai_harness.step_persistence import StepPersistence, InMemoryStepStore

store = InMemoryStepStore()
librarian = Agent(
    'openai:gpt-5',
    capabilities=[StepPersistence(store=store, agent_name='code_librarian')],
)


async def main():
    await librarian.run('Find ThinkingPartDelta and confirm the callable allowance')


asyncio.run(main())

That is the whole setup. run_id is always per-Agent.run call, matching pydantic_ai’s RunContext.run_id. For multi-turn logical grouping use conversation_id= — that is the pydantic_ai-native primitive for it (see Three-level identity).

run_id resolution per call:

Explicit run_id='libr-1' becomes the id for this one call. This suits single-shot use cases (a deterministic id for testing, replay, debugging, or a one-off scripted run). Reusing one capability instance with the same explicit run_id across multiple .run() calls raises ValueError in before_run. The tool-effect ledger is keyed by (run_id, tool_call_id) and providers reuse deterministic tool-call ids, so a silent collision would erase the unknown_after_crash signal. Use conversation_id= for multi-turn grouping instead.
agent_name set, run_id unset derives '{agent_name}-{8-char-hex}', freshly materialised in for_run per .run() call. Reusing one capability instance across runs yields distinct ids (code_librarian-a3b2, code_librarian-c9d1, and so on). This is the recommended default for delegate capabilities.
Neither set falls back to ctx.run_id (pydantic_ai’s auto-generated id) per .run() call, and to a UUID4 if that is absent.

The orchestrator pattern — one logical agent serving many turns — uses conversation_id, not a shared run_id:

import asyncio

from pydantic_ai import Agent
from pydantic_ai_harness.step_persistence import StepPersistence, InMemoryStepStore

store = InMemoryStepStore()
orchestrator = Agent(
    'openai:gpt-5',
    capabilities=[StepPersistence(store=store, agent_name='orchestrator')],
)


async def main():
    for turn in turns:
        await orchestrator.run(turn, conversation_id='orch-conv')

    # All turns of this orchestrator, chronological:
    records = await store.list_runs(conversation_id='orch-conv')


asyncio.run(main())

Three-level identity

The capability mirrors pydantic_ai’s identity stack:

Concept	Definition	Granularity
`conversation_id`	The dialogue. Resolved by pydantic_ai from the `conversation_id=` argument to `Agent.run`, or the most recent `conversation_id` on `message_history`, or a fresh UUID7.	sequence of runs
`run_id`	One `Agent.run` invocation.	one step in the sequence
`step_index`	Graph-node count within a run (`ctx.run_step`).	one node within one run

StepEvent.conversation_id and RunRecord.conversation_id are populated from ctx.conversation_id. So three .run() calls sharing one conversation_id produce three distinct run_ids, all queryable as a group:

import asyncio


async def main():
    runs = await store.list_runs(conversation_id='conv-abc')  # 3 records, chronological


asyncio.run(main())

Continuing a delegate’s investigation

pydantic_ai already has message_history= for “carry on with this prior context”. StepPersistence does not introduce a parallel mechanism. It exposes one helper that loads the most recent settled snapshot:

import asyncio

from pydantic_ai import Agent
from pydantic_ai_harness.step_persistence import (
    StepPersistence,
    InMemoryStepStore,
    continue_run,
)

store = InMemoryStepStore()
librarian = Agent(
    'openai:gpt-5',
    capabilities=[StepPersistence(store=store, agent_name='code_librarian')],
)


async def main():
    # Earlier: tag the first turn with a conversation id so the follow-up can find it.
    await librarian.run(
        'Find ThinkingPartDelta and confirm the callable allowance',
        conversation_id='libr-conv',
    )

    # Later (possibly a different process):
    prior_run = (await store.list_runs(conversation_id='libr-conv'))[-1].run_id
    history = await continue_run(store, run_id=prior_run)
    await librarian.run(
        'Read _apply_provider_details_delta and check the path',
        message_history=history,
        conversation_id='libr-conv',   # keep the conversation grouping
    )


asyncio.run(main())

fork_run(store, run_id=...) returns the same shape but is intended when the caller wants a branched logical run from that snapshot point (the new run gets a fresh run_id and probably a fresh conversation_id).

What “safe to continue from” means

By default continue_run returns the messages of the latest complete snapshot for that run_id — a point whose tool work was fully settled when captured. Snapshots are written at these boundaries:

after every CallToolsNode whose tool calls all returned — the pending tool-return request is folded in, so the point is durable the moment the tool completes, before the next model request is even sent,
at after_run, when the run ended past that boundary (a run that reached no boundary at all, or an Agent.run_stream whose closing response lands after the last one), and
when a run fails: the live history at failure time is saved, whatever its shape — a model request that raises after a clean tool cycle produces a complete snapshot; a crash mid-tool-cycle produces an interrupted one carrying every completed cycle.

An interrupted snapshot is sendable on resume — pydantic-ai (>= 2.10) repairs broken tool-call/result pairing before every model request — but not necessarily safe: a pending tool call may be re-executed (resuming without a new prompt) or closed out with a synthesized interrupted return, and neither says whether the original side effect happened. That is the tool-effect ledger’s job. So the default read path skips interrupted snapshots; pass include_interrupted=True to continue_run / fork_run / latest_snapshot after checking list_unresolved_tool_effects. If no matching snapshot exists, continue_run raises LookupError.

Run lineage: `parent_run_id`

parent_run_id is a lineage label, not a functional dependency. It does two things:

Every StepEvent and RunRecord carries it, so you can filter and group.
store.list_runs(parent_run_id='orch-1') returns every delegate run pointing at that orchestrator.

It is auto-inferred for in-process delegation: when an orchestrator’s tool synchronously calls a delegate’s Agent.run(...), the delegate’s StepPersistence picks up the orchestrator’s run_id via a ContextVar that the orchestrator’s wrap_run set. No threading required:

import asyncio

from pydantic_ai import Agent
from pydantic_ai_harness.step_persistence import StepPersistence, InMemoryStepStore

store = InMemoryStepStore()
orchestrator = Agent(
    'openai:gpt-5',
    capabilities=[StepPersistence(store=store, agent_name='orchestrator')],
)
librarian = Agent(
    'openai:gpt-5',
    capabilities=[StepPersistence(store=store, agent_name='code_librarian')],
)


@orchestrator.tool_plain
async def ask_librarian(question: str) -> str:
    result = await librarian.run(question)   # parent_run_id auto-filled
    return result.output


async def main():
    # Tag the orchestrator turn so the lookup below can find its run_id.
    await orchestrator.run(
        'Where is ThinkingPartDelta defined?',
        conversation_id='orch-conv',
    )

    # All librarian runs now point at the orchestrator's run_id:
    orch_run_id = (await store.list_runs(conversation_id='orch-conv'))[-1].run_id
    delegates = await store.list_runs(parent_run_id=orch_run_id)


asyncio.run(main())

Set parent_run_id= explicitly to override (for example, cross-process delegation where ContextVars do not propagate).

parent_run_id is distinct from conversation_id. The orchestrator and delegate usually live in different conversations (the orchestrator talks to a user; the delegate talks to itself). But they share a parent-child link.

Inspecting a run tree

list_runs returns matches sorted by started_at ascending across all backends — pick the most recent with [-1].

import asyncio


async def main():
    # Every delegate of one orchestrator run (chronological)
    delegates = await store.list_runs(parent_run_id='orch-3f2a')

    # Every run in one dialogue (multi-turn conversation across many .run() calls)
    turns = await store.list_runs(conversation_id='conv-abc')
    latest_turn = turns[-1]

    # Filters combine (AND):
    focused = await store.list_runs(
        parent_run_id='orch-3f2a',
        conversation_id='libr-conv',
    )

    # Detail per run:
    events = await store.list_events(run_id=delegates[0].run_id)
    snapshot = await store.latest_snapshot(run_id=delegates[0].run_id)
    unresolved = await store.list_unresolved_tool_effects(run_id=delegates[0].run_id)


asyncio.run(main())

Failure recovery

import asyncio


async def main():
    # An earlier delegate run died mid-investigation.
    events = await store.list_events(run_id='libr-3f2a')
    unresolved = await store.list_unresolved_tool_effects(run_id='libr-3f2a')
    for record in unresolved:
        # status == 'started' with no terminal update -- unknown_after_crash.
        print(f'tool {record.tool_name} ({record.tool_call_id}) may or may not have run')
        print(f'  idempotency_key={record.idempotency_key}  '
              f'effect_summary={record.effect_summary}')

    # Decide whether to resume or branch:
    history = await continue_run(store, run_id='libr-3f2a')
    # If the unresolved tools were read-only and safe to redo:
    await librarian.run('continue investigating', message_history=history,
                        conversation_id='libr-conv')
    # If side effects might have happened and the orchestrator wants a fresh attempt:
    history = await fork_run(store, run_id='libr-3f2a')
    # ... pass to a new delegate run with a different agent_name / conversation_id.

    # To resume from the interrupted frontier itself (the crashed cycle included),
    # after checking the unresolved effects above:
    history = await continue_run(store, run_id='libr-3f2a', include_interrupted=True)


asyncio.run(main())

Side-effect deduplication is the orchestrator’s responsibility. Tools that write external state should annotate their in-flight ToolEffectRecord via annotate_tool_effect:

from pydantic_ai import RunContext
from pydantic_ai_harness.step_persistence import annotate_tool_effect


@orchestrator.tool
async def set_label(ctx: RunContext[Deps], issue: int, label: str) -> str:
    await annotate_tool_effect(
        store,
        ctx,
        idempotency_key=f'issue-{issue}::label::{label}',
        effect_summary=f'set label {label!r} on issue #{issue}',
    )
    await github.set_label(issue, label)   # the actual side effect
    return 'ok'

The helper reads the active run_id from the StepPersistence ContextVar and tool_call_id / tool_name from ctx, then merges the metadata into the prior record. It is a no-op when called outside a step-persistence-wrapped tool call. after_tool_execute preserves both fields when it writes the terminal completed / failed entry.

Compaction receipt handles

StepPersistence.compaction_transcript_handle() exposes the current run_id to compaction receipts. It is an identifier for this store’s persisted run history, not a promise that the pre-compaction transcript remains available: snapshots can already contain compacted history and configured retention can delete older snapshots.

Backends

InMemoryStepStore — process-local; great for tests.
FileStepStore(directory) — directory layout under <directory>/<run_id>/:
- run.json — RunRecord (lineage)
- events.jsonl — append-only StepEvents
- tool_effects.jsonl — append-only ToolEffectRecords, scoped to this run
- snapshots/{seq}.json — ContinuableSnapshots, named by a per-run monotonic counter (not step_index, which would collide when the same run_id is reused across Agent.run calls, since ctx.run_step resets to 0 each call).
SqliteStepStore(database='runs.db') — single SQLite file with tables runs, events, snapshots, tool_effects, and a sibling media table for externalized blobs (see Persisting media below). WAL mode is enabled; tool_effects upserts per (run_id, tool_call_id) so the latest state wins; snapshots use AUTOINCREMENT seq to mirror FileStepStore._next_snapshot_seq. Databases created before the snapshot state column existed gain it automatically on open (existing rows read as complete). Pass connection= instead of database= to share a sqlite3.Connection with the rest of your application; the connection must be opened with check_same_thread=False because hook calls are dispatched onto a worker thread.
MongoStepStore(client= or db_url=, database=...) — MongoDB collections runs, events, snapshots, tool_effects, and counters (atomic $inc allocates the monotonic seq). runs._id = run_id enforces the single-shot run_id contract. Needs the mongodb extra (pip install pydantic-ai-harness[mongodb], which installs pymongo>=4.17.0); pass a shared AsyncMongoClient as client=, or a connection string as db_url= (the store then owns the client — call await store.aclose() to release it). Individual parts at or above media_threshold_bytes externalize by default to a MongoMediaStore on the same client. That is a per-value offload, not an aggregate cap: a snapshot of many below-threshold parts can still exceed MongoDB’s 16 MiB document limit and fail on insert, so lower the threshold if that is a risk for your workload.

All implement the same async StepStore protocol, so capability hooks never block the event loop on the file/sqlite backends (I/O is dispatched via anyio.to_thread); the Mongo backend is natively async.

FileStepStore validates run_id against [A-Za-z0-9_.-]{1,200} (and rejects ..) to prevent path traversal. Callers passing user-controlled IDs should still sanitise first.

What `MongoStepStore` creates on first write

The store issues createIndex on its first write, for eight indexes: conversation_id and parent_run_id (both sparse) plus started_at on runs; (run_id, seq) on events; (run_id, seq) and (run_id, state, seq) on snapshots; and a unique (run_id, tool_call_id) plus (run_id, status) on tool_effects. Its default MongoMediaStore adds one more, described on the media page. Three consequences worth knowing before pointing the store at an existing deployment:

The connecting user needs the privilege to create indexes. A restricted Atlas role without it fails on the first write, not at construction.
The unique index build fails if an existing tool_effects collection already holds duplicate (run_id, tool_call_id) pairs.
Index builds against already-populated collections cost time and I/O on that first call.

RunRecord.metadata and StepEvent.metadata are stored as nested documents, so their keys become BSON field names: keys containing . or starting with “RunRecord.metadataandStepEvent.metadataare stored as nested documents, so their keys become BSON field names: keys containing.or starting with need [MongoDB 5.0 or later](https://www.mongodb.com/docs/manual/core/dot-dollar-considerations/), and a key containing a NULL byte is rejected by the BSON encoder before it reaches the server. CI exercises both Mongo backends againstmongo:8`.

Bounding snapshot growth

Each step writes a new full-history snapshot keyed by an incrementing seq, and nothing is pruned by default. Within one long Agent.run the snapshot count equals the number of settled tool-call steps, so a long single run pays a growing storage cost.

All four stores — InMemoryStepStore, FileStepStore, SqliteStepStore, and MongoStepStore — accept an opt-in max_snapshots_per_run: int | None (default None, unbounded — byte-for-byte the prior behavior). When set to N >= 1, each save_snapshot prunes the run down to a retain set:

the newest N snapshots by seq,
the newest snapshot overall (serves latest_snapshot(include_interrupted=True)),
the newest complete snapshot (serves the default read path).

The last two keep both read modes correct even when the newest N snapshots are all interrupted and the newest resumable complete sits below that window, so the retain set can exceed N. from_spec(..., max_snapshots_per_run=N) forwards the bound to the store it constructs (backend='memory', 'file', or 'sqlite'; a Mongo store is built directly, not from a spec).

from pydantic_ai_harness.step_persistence import FileStepStore

store = FileStepStore('runs', max_snapshots_per_run=8)

Pruning a snapshot never deletes its externalized media: blobs are content-addressed and may be shared across snapshots and runs, so orphaned-blob GC is out of scope (see the non-goals below). Age-based (TTL) expiry is out of scope too — it belongs at whole-run granularity, not per snapshot.

Bounded retention discards older per-step snapshots, including pre-compaction ones. Any downstream that reconstructs history by unioning a run’s retained snapshots — snapshot search or a compaction receipt keyed on run_id — can only see what is retained. With a tight bound (for example max_snapshots_per_run=1) the older, pre-compaction states are gone, so treat the bound as a hard limit on how far back such recovery can reach. Leave the bound at None, or set it high enough to cover the history you need to recover, when historical reconstruction matters.

Persisting media

BinaryContent payloads (images, audio, documents, video) inlined as base64 inside a snapshot would balloon every file or row containing the message; a large text part (e.g. a big tool-return string) does the same and can push a MongoStepStore snapshot past MongoDB’s 16 MiB document cap (#440). The file, sqlite, and mongo backends externalize any BinaryContent.data, and any part whose string content is at or above 64 KiB, through a configured MediaStore, leaving a URI reference in the snapshot. The same media_threshold_bytes governs binary and text alike; there is no separate text knob. Round-trip is transparent: latest_snapshot(...).messages[*] returns the original BinaryContent bytes and text.

Text externalization is not Mongo-only and has no opt-out short of media_store=None: the walker is shared, so an existing FileStepStore or SqliteStepStore deployment starts writing blobs for large text parts as well as binary ones from this release on. Snapshots written before it still restore — the reader recognises the older binary marker shape. This compatibility is upgrade-only: a release that predates text externalization treats every marker as binary, so it cannot validate a snapshot containing an externalized text marker. Keep a current reader for persisted snapshots that contain those markers.

StepStore	Default `media_store`	Where blobs live
`InMemoryStepStore`	not applicable	bytes stay in the in-memory snapshot
`FileStepStore`	`DiskMediaStore(<root>/media/)`	`<root>/media/<sha256>.bin`
`SqliteStepStore`	`SqliteMediaStore(database=<same db>)`	sibling `media` table in the same DB
`MongoStepStore`	`MongoMediaStore(client=<same client>)`	sibling `media` + `media_chunks` collections

Override the destination by passing your own MediaStore:

from pydantic_ai_harness.step_persistence import FileStepStore
from pydantic_ai_harness.media import S3MediaStore

store = FileStepStore(
    'runs',
    media_store=S3MediaStore(
        bucket='my-bucket',
        endpoint='https://<account>.r2.cloudflarestorage.com',
        region='auto',
        access_key_id=...,
        secret_access_key=...,
    ),
    media_threshold_bytes=64 * 1024,  # raise or lower if you want
)

Opt out entirely (keep bytes inline in the snapshot JSON/row):

from pydantic_ai_harness.step_persistence import FileStepStore, SqliteStepStore

FileStepStore('runs', media_store=None)
SqliteStepStore(database='runs.db', media_store=None)

URIs are media+sha256://<hex>, content-addressed. The same blob written through any MediaStore resolves the same way, so dedup is automatic and moving the underlying storage is a one-line swap. The shipped implementations are:

DiskMediaStore(directory) — one file per blob at <directory>/<sha256>.bin.
SqliteMediaStore(database=...) or SqliteMediaStore(connection=...) — one row per blob (INSERT OR IGNORE for content-addressed dedup).
S3MediaStore(bucket=, endpoint=, region=, access_key_id=, secret_access_key=) — path-style URLs plus handrolled SigV4. Compatible with AWS S3, Cloudflare R2 (region='auto'), MinIO, and other S3-compatible providers. PUT/GET/HEAD only — no multipart, lifecycle, or listing in v1.
MongoMediaStore(client= or db_url=, database=...) — MongoDB, needs the mongodb extra. Each blob is sha256-addressed chunks across a media manifest document and a sibling media_chunks collection (manual chunking rather than GridFS, so dedup is preserved — see the media page), so a blob larger than one BSON document still stores and reads back. Chunking bounds the document, not memory: there is no streaming API, so each blob is held whole in process memory on both put and get. The manifest holds MediaContext.metadata inline and is not chunked, so keep per-blob metadata small. collection= renames both collections and chunk_size_bytes= (default 8 MiB) sets the split size.

Exposing externalized bytes as URLs

Each store accepts a public_url= callable that turns the canonical media+sha256://<hex> URI into a URL the model can fetch directly. The forthcoming MediaExternalizer capability will use this to swap BinaryContent parts for ImageUrl / AudioUrl / other URL parts before the model sees the message, letting providers fetch big media over the wire without re-encoding bytes into the request body.

Static base URL (public R2 bucket, CDN):

from pydantic_ai_harness.media import S3MediaStore, make_static_public_url

store = S3MediaStore(
    bucket='my-bucket',
    endpoint='https://<acc>.r2.cloudflarestorage.com',
    region='auto',
    access_key_id=..., secret_access_key=...,
    key_prefix='media/',
    public_url=make_static_public_url('https://pub-abc.r2.dev', key_prefix='media/'),
)

Presigned or rotating-signature URL — pass any async callable that takes (uri, MediaContext):

from pydantic_ai_harness.media import MediaContext, S3MediaStore


async def presign(uri: str, ctx: MediaContext) -> str:
    key = 'media/' + uri.removeprefix('media+sha256://') + '.bin'
    return await my_signer.generate(key, ttl=3600, content_type=ctx.media_type)


store = S3MediaStore(..., public_url=presign)

`MediaContext`, an extensible per-operation bag

Every MediaStore method (put, get, exists, public_url, get_metadata) and both user-supplied callables (PublicUrlResolver, KeyStrategy) accept a MediaContext:

from collections.abc import Mapping
from dataclasses import dataclass, field


@dataclass(frozen=True, kw_only=True)
class MediaContext:
    media_type: str | None = None                    # e.g. 'image/png'
    filename: str | None = None                      # original filename, when known
    metadata: Mapping[str, str] = field(default_factory=dict)  # user-supplied tags

All fields default; new fields are added non-breakingly as use cases emerge. Pass what you have, ignore the rest.

Persistence by store. get_metadata(uri) round-trips the user-supplied metadata mapping on all four stores. media_type is also persisted but is not part of what get_metadata returns (it is stored for the byte payload itself, for example as the Content-Type).

SqliteMediaStore writes metadata to a JSON column and media_type to a dedicated column.
S3MediaStore sends metadata as signed x-amz-meta-* headers (ASCII alphanumeric plus dash key names) and media_type as Content-Type; get_metadata reads the x-amz-meta-* values back from the HEAD response.
DiskMediaStore writes a sidecar JSON file (<resolved>.meta.json) alongside each blob, atomic via tmp plus rename. Sidecars are absent only when the put carried no metadata.
MongoMediaStore writes metadata as a JSON string and media_type as a dedicated field on the blob’s manifest document (the media collection by default); get_metadata decodes the JSON string back. Because the mapping is one JSON string rather than nested fields, metadata keys are not subject to BSON field-name rules here.

`key_strategy`: controlling the backend storage path

Default is <sha256>.bin. DiskMediaStore and S3MediaStore accept overrides to fit existing layouts; SqliteMediaStore and MongoMediaStore do not (the digest is their primary key, so a user-chosen key would either break dedup or be a no-op — use table= / collection= to move the rows or documents):

from pydantic_ai_harness.media import DiskMediaStore, MediaContext


def by_media_type(uri: str, ctx: MediaContext) -> str:
    digest = uri.removeprefix('media+sha256://')
    ext = {'image/png': '.png', 'image/jpeg': '.jpg'}.get(ctx.media_type or '', '.bin')
    return f'images/{digest}{ext}'


store = DiskMediaStore('runs', key_strategy=by_media_type)

Caveat: if your strategy depends on context.media_type (for example, to pick an extension), get(uri) and exists(uri) will not find the blob unless the same context is supplied at read time. For pure path-organisation strategies (no context dependency) the constraint does not apply.

DiskMediaStore rejects strategies that produce absolute paths or paths containing .. segments, to prevent escaping the store directory.

Separately, all four stores accept a public_url= resolver, useful when a CDN, local HTTP server, or signed-URL service fronts the bytes. Without it public_url(...) returns None (the model never sees a URL unless a resolver is configured and it returns a string).

pydantic_ai providers transparently download bytes from a URL when the target model does not natively accept that URL type, so emitting a URL is always safe: you only ever lose wire savings, never correctness.

Persisting to unsupported backends

DynamoDB, Postgres, Redis, GCS, and other backends are out of scope for this release. Write your own StepStore (about ten methods on a Protocol) or your own MediaStore (five methods: put, get, exists, public_url, get_metadata) and pass it via store= / media_store=. Please open an issue if you ship one — we want to feed the eventual shared adapter layer with N >= 3 real implementations before abstracting.

What this capability does not do

It does not restore capability per-run state, graph-node state, retry counters, or in-flight streaming responses.
It does not deduplicate replayed side effects automatically. Tools that write artifacts, labels, PRs, or external state should call annotate_tool_effect(store, ctx, ...) (see Failure recovery) so the orchestrator can decide whether replay is safe.
It does not prune events, and by default does not prune snapshots. Retention is the caller’s responsibility; snapshot growth can be bounded opt-in with max_snapshots_per_run (see Bounding snapshot growth).
It does not garbage-collect externalized media. Pruning a snapshot leaves its content-addressed blobs in place, since they may be shared across snapshots and runs.
It does not emit OpenTelemetry spans. pydantic_ai’s Instrumentation capability already spans agent run / chat / running tool and populates gen_ai.agent.name, gen_ai.agent.call.id, gen_ai.conversation.id via baggage. A future change may add step-persistence attributes to the active span; that is tracked as a follow-up issue.

API reference

StepPersistence

Bases: AbstractCapability[AgentDepsT]

Append-only step log + continuable snapshots + tool-effect ledger.

The capability emits a StepEvent at every interesting boundary (run/model-request/tool-call start, completion, failure), records a ToolEffectRecord per tool call so the orchestrator can decide whether replay is safe, and saves a ContinuableSnapshot at every settled CallToolsNode boundary — folding in the pending tool-return request, so the point is durable the moment the tool completes — plus a fallback save at after_run when the run ends past that boundary. A run that fails saves the live at-failure history (see on_run_error), classified by its tool-work state: complete when every tool call is resolved, interrupted otherwise.

A run that crashes between before_tool_execute and after_tool_execute leaves a visible event trail, a started tool-effect record (the unknown_after_crash signal), and an interrupted snapshot carrying every completed cycle. The default latest_snapshot / continue_run read path only returns complete snapshots; pass include_interrupted=True to resume from the interrupted frontier after consulting list_unresolved_tool_effects.

from pydantic_ai import Agent
from pydantic_ai_harness.step_persistence import StepPersistence, InMemoryStepStore

store = InMemoryStepStore()
librarian = Agent(
    'openai:gpt-5',
    capabilities=[StepPersistence(store=store, agent_name='code_librarian')],
)
await librarian.run('Find ThinkingPartDelta and confirm the callable allowance')

Use continue_run(store, run_id=...) / fork_run(store, run_id=...) to load a prior snapshot, then pass the result to Agent.run(..., message_history=...).

Attributes

store

Backend that records events, snapshots, and tool effects.

Type: StepStore Default: field(default_factory=InMemoryStepStore)

agent_name

Logical agent name (e.g. code_librarian, reproducer).

Used as a stable prefix for the auto-derived run_id so store inspection shows readable IDs like code_librarian-a3b2.

Type: str | None Default: None

run_id

Identifier for this one Agent.run call.

run_id is per-call, matching pydantic_ai.RunContext.run_id. For multi-turn logical grouping use conversation_id on Agent.run(...) — that is the pyai-native primitive for it.

Resolution order (materialised in for_run):

Explicit value → used as-is. Single-shot use cases: deterministic id for testing, replay, debugging. Reusing the capability across multiple .run() calls with the same explicit run_id raises ValueError in before_run — the tool-effect ledger keys on (run_id, tool_call_id) and providers reuse deterministic tool-call ids, so a silent collision would erase the unknown_after_crash signal. Use conversation_id= on Agent.run for multi-turn grouping.
agent_name set, run_id unset → {agent_name}-{short-uuid}, freshly materialised per .run(). Reusing the capability instance yields distinct ids. Recommended default for delegate capabilities.
Neither set → ctx.run_id per .run(), falling back to UUID4.

Type: str | None Default: None

parent_run_id

Run that spawned this one.

Auto-inferred from the enclosing StepPersistence wrap_run scope — when an orchestrator’s tool synchronously calls a delegate’s Agent.run(...), the delegate picks up the orchestrator’s run_id here without manual threading. Set explicitly to override (e.g. for cross-process delegation where ContextVars do not propagate).

Type: str | None Default: None

metadata

Free-form metadata stored on the RunRecord and on each event.

Type: dict[str, str] Default: field(default_factory=_empty_metadata)

Methods

from_spec

@classmethod

def from_spec(cls, *args: Any, **kwargs: Any) -> StepPersistence[Any]

Construct from a serialised spec.

Supports backend='memory' (default), backend='file' (with directory), or backend='sqlite' (with database). Raises ValueError for any other backend value — silently falling back to in-memory storage would turn a typo into accidental non-durability.

max_snapshots_per_run (default None, unbounded) is forwarded to the constructed store to bound per-run snapshot growth.

Returns

StepPersistence[Any]

compaction_transcript_handle

def compaction_transcript_handle() -> str | None

Retrieval handle to this run’s transcript, for compaction receipts.

Satisfies the compaction TranscriptHandleProvider protocol structurally (no import coupling). A compaction strategy discovers this capability via RunContext.capabilities and records its run id. Returns None before for_run has materialised the id.

Returns

str | None

for_run

@async

def for_run(ctx: RunContext[AgentDepsT]) -> AbstractCapability[AgentDepsT]

Materialise run_id and parent_run_id for this Agent.run call.

Reads the contextvar set by any enclosing StepPersistence.wrap_run before the local run overwrites it, so a delegate’s parent_run_id ends up pointing at its orchestrator’s run_id.

A separate ContextVar is needed because pydantic_ai’s own cross-run signals (RUN_ID_BAGGAGE_KEY via OTel baggage, RunContext.run_id, and _CURRENT_RUN_CONTEXT) are single-slot: the inner Instrumentation.wrap_run overwrites them before any nested capability sees the parent. The harness-local contextvar lets us snapshot the parent here, before the local wrap_run rebinds it.

Returns

AbstractCapability[AgentDepsT]

wrap_run

@async

def wrap_run(
    ctx: RunContext[AgentDepsT],
    *,
    handler: WrapRunHandler,
) -> AgentRunResult[Any]

Push this run’s id onto the contextvar so nested delegates can read it.

Returns

AgentRunResult[Any]

before_run

@async

def before_run(ctx: RunContext[AgentDepsT]) -> None

When the caller pinned an explicit run_id, reject reuse — the tool-effect ledger keys on (run_id, tool_call_id) and providers reuse deterministic tool-call ids, so a second Agent.run with the same explicit run_id would silently collide. The auto-derived cases cannot trigger this check because each call materialises a fresh id in for_run.

Returns

None

after_run

@async

def after_run(
    ctx: RunContext[AgentDepsT],
    *,
    result: AgentRunResult[Any],
) -> AgentRunResult[Any]

Emit run_completed, saving a final snapshot only as a fallback.

When a terminal CallToolsNode already saved the final history via after_node_run it carries the correct step_index, whereas by after_run ctx.run_step is reset to 0 — so re-saving would both duplicate the tail and stamp a misleading step_index. We save only when the run ended past the newest boundary snapshot.

That covers a run which reached no provider-valid boundary at all, and Agent.run_stream, which ends through SetFinalResult rather than a terminal CallToolsNode and appends its closing response after the last boundary — leaving after_run the only hook that sees the full run.

Returns

AgentRunResult[Any]

on_run_error

@async

def on_run_error(
    ctx: RunContext[AgentDepsT],
    *,
    error: BaseException,
) -> AgentRunResult[Any]

Persist the live at-failure history as the run’s last resume point, then emit run_failed.

The single error-path save site: reads the list reference stashed by after_node_run (see _stash_live_history), whose content at this point is the full history the run had built when it failed — including a failing model request’s payload and any partial tool returns captured by the graph during unwind. Nothing is compared against the store: the live history is by definition the newest state, so an earlier boundary snapshot is simply superseded, and a history a sticky processor trimmed is persisted as trimmed — exactly what the next request would have sent.

The history is saved whenever it contains a model response (a bare prompt equals restarting the run), classified complete when every tool call is resolved and interrupted otherwise. Interrupted snapshots stay off the default latest_snapshot read path.

Returns

AgentRunResult[Any]

on_model_request_error

@async

def on_model_request_error(
    ctx: RunContext[AgentDepsT],
    *,
    request_context: ModelRequestContext,
    error: Exception,
) -> ModelResponse

Emit model_request_failed and re-raise.

No snapshot is saved here: the failing request’s payload already sits in the live history (the graph appends the request before sending), so on_run_error’s save covers it. A failure the model layer recovers from (retry, fallback) needs no rescue at all.

Returns

ModelResponse

after_node_run

@async

def after_node_run(
    ctx: RunContext[AgentDepsT],
    *,
    node: AgentNode[AgentDepsT],
    result: NodeResult[AgentDepsT],
) -> NodeResult[AgentDepsT]

Save a continuable snapshot after a settled CallToolsNode, and refresh the live-history stash.

At that boundary every tool call from the preceding ModelRequestNode has a matching tool return, so the history is provider-valid. The returned ModelRequestNode carries those returns and is not yet in ctx.messages, so its request is folded in before validation — without it a worker killed right after a completed tool call would leave no resume point at all (#373). is_provider_valid doubles as a defense in case a custom node reshapes history, and the saved count goes to snapshot_saved so after_run can tell whether the run ended past this boundary.

This save is the durable one: it lands in the store while the run is still healthy, so it survives a hard kill that fires no hook. The error path (on_run_error) only rescues histories that a raise unwinds through.

Every node boundary also re-stashes the live message list so that on_run_error can persist the at-failure history when a later node raises before its own after_node_run fires. The stash holds that list by reference, so the snapshot candidate rebinds to a new list rather than appending to it — an append would leak result.request into the history the error path later reads, duplicating it once the graph appends the request itself.

Returns

NodeResult[AgentDepsT]

Step Persistence

What it gives you

Quick start

Three-level identity

Continuing a delegate’s investigation

What “safe to continue from” means

Run lineage: parent_run_id

Inspecting a run tree

Failure recovery

Compaction receipt handles

Backends

What MongoStepStore creates on first write

Bounding snapshot growth

Persisting media

Exposing externalized bytes as URLs

MediaContext, an extensible per-operation bag

key_strategy: controlling the backend storage path

Persisting to unsupported backends

What this capability does not do

Related

API reference

StepPersistence

Attributes

store

agent_name

run_id

parent_run_id

metadata

Methods

from_spec

Returns

compaction_transcript_handle

Returns

for_run

Returns

wrap_run

Returns

before_run

Returns

after_run

Returns

on_run_error

Returns

on_model_request_error

Returns

after_node_run

Returns

Run lineage: `parent_run_id`

What `MongoStepStore` creates on first write

`MediaContext`, an extensible per-operation bag

`key_strategy`: controlling the backend storage path