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 only at boundaries where the message history is provider-valid: every ToolCallPart has a matching ToolReturnPart or RetryPromptPart, with no orphan, duplicate, or out-of-order returns. 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 provider-valid 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

continue_run only returns the messages of the latest provider-valid snapshot for that run_id. Snapshots are written at two boundaries:

after every CallToolsNode completes (all tool calls returned), and
at after_run, as a fallback if the run reached no such boundary.

A run that crashed mid-tool-call has events (tool_call_started) but no snapshot for that point. continue_run returns the snapshot from the previous safe boundary, not the failed step. If no continuable snapshot exists at all, 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.


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.

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. 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.

All three 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).

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.

Persisting media

BinaryContent payloads (images, audio, documents, video) inlined as base64 inside a snapshot would balloon every file or row containing the message. Both FileStepStore and SqliteStepStore externalize any BinaryContent.data at or above 64 KiB through a configured MediaStore, leaving a URI reference in the snapshot. Round-trip is transparent: latest_snapshot(...).messages[*] returns BinaryContent with the original bytes.

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

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.

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 three 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.

`key_strategy`: controlling the backend storage path

Default is <sha256>.bin. DiskMediaStore and S3MediaStore accept overrides to fit existing layouts; SqliteMediaStore does not (its primary key is the digest, so a user-chosen key would either break dedup or be a no-op):

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 three 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 (three methods) 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 clean up old snapshots or events. Retention is the caller’s responsibility.
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 provider-valid boundary — the end of each CallToolsNode — plus a fallback save at after_run if the run reached no such boundary.

A run that crashes between before_tool_execute and after_tool_execute leaves a visible event trail and a started tool-effect record, but no new continuable snapshot — the latest snapshot reflects the last provider-valid state.

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.

Returns

StepPersistence[Any]

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.

The terminal CallToolsNode already saved the final provider-valid snapshot via after_node_run, carrying the correct step_index. By after_run, ctx.run_step is reset to 0, so re-saving here would both duplicate the tail and stamp a misleading step_index. We only save when the run produced no snapshot at all (no provider-valid node boundary was reached), as a last-resort capture of the final state.

Returns

AgentRunResult[Any]

on_run_error

@async

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

Emit run_failed so a killed run leaves a visible event trail.

Returns

AgentRunResult[Any]

after_node_run

@async

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

Save a mid-run continuable snapshot after CallToolsNode succeeds.

At that boundary every tool call from the preceding ModelRequestNode has a matching tool return, so the history is provider-valid. Snapshots are filtered through is_provider_valid defensively in case a custom node reshapes history.

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

Backends

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

for_run

Returns

wrap_run

Returns

before_run

Returns

after_run

Returns

on_run_error

Returns

after_node_run

Returns

Run lineage: `parent_run_id`

`MediaContext`, an extensible per-operation bag

`key_strategy`: controlling the backend storage path