Step Persistence

이 문서에서는 StepPersistence capability를 소개해요. 에이전트가 각 경계에서 무엇을 했는지, 실행을 안전하게 재개할 수 있는지와 별개로 기록해요. 이것은 하위 에이전트에 위임하는 오케스트레이터를 위한 지속성 기반이에요. 예를 들어 AICA 오케스트레이터가 code_librarian을 띄워 한 기호를 조사한 뒤, 후속 질문으로 그 위임의 조사를 계속하는 경우가 있어요.

출처: 문서

본문

이것은 완전한 그래프 상태 체크포인트가 아니에요. capability 상태 복원, 워크스페이스 스냅샷, 그래프 노드 재개는 범위 밖이며 별도로 추적돼요(pydantic-ai-harness 이슈 #149, #196 참고).

Pydantic AI Harness가 0.x 릴리스인 동안 API는 마이너 릴리스 사이에 바뀔 수 있어요. 바뀌면 deprecation 경고와 릴리스 노트 마이그레이션 안내가 정확히 어떻게 업그레이드할지 알려줘요. 버전 정책을 참고하세요.

무엇을 주는가 (What it gives you)

추가 전용 단계 이벤트 (Append-only step events). 모든 흥미로운 경계(실행 시작/끝, 모델 요청, 도구 호출, 실패)가 StepEvent를 추가해요. 도구 호출 중간에 죽은 실행도 사용 가능한 이벤트 흔적을 남겨요.
재개 가능한 스냅샷 (Continuable snapshots). ContinuableSnapshot은 정착된 노드 경계에서 저장되고, 실패한 실행은 실패 시점의 살아있는 히스토리를 저장해요. 각 스냅샷은 state를 지녀요. 모든 ToolCallPart에 일치하는 결과가 있으면 complete, 캡처가 미정산 도구 작업(예: 도구 주기 중간 크래시)을 담고 있으면 interrupted예요. latest_snapshot과 continue_run은 호출자가 include_interrupted=True를 넘기지 않으면 complete 스냅샷만 반환해요. 스냅샷의 messages를 Agent.run(message_history=...)에 다시 넘겨 계속하거나 포크하세요.
도구 효과 원장 (Tool-effect ledger). 모든 도구 호출의 수명 주기(started, completed, failed)가 (run_id, tool_call_id)에 대해 기록돼요. 크래시 후 started 기록이 있고 종료 업데이트가 없는 도구는 unknown_after_crash로 취급해야 해요. 부작용이 일어났을 수도, 아닐 수도 있기 때문이에요.
계보 메타데이터 (Lineage metadata). conversation_id(시퀀스)와 parent_run_id(계층)는 독립적인 축이에요. 3단계 신원 참고.

빠른 시작 (Quick start)

import asyncio

from pydantic_ai import Agent
from pydantic_ai_harness import StepPersistence
from pydantic_ai_harness.step_persistence import 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())

이것이 전체 설정이에요. run_id는 항상 Agent.run 호출별이며, pydantic_ai의 RunContext.run_id와 일치해요. 다중 턴 논리 그룹에는 conversation_id=를 사용하세요. 그것이 pydantic_ai 네이티브 원시형이에요(3단계 신원 참고).

호출별 run_id 해석:

명시적 run_id='libr-1' 은 이 호출 하나의 id가 돼요. 단일 샷 사용 사례(테스트·재생·디버깅·일회성 스크립트 실행을 위한 결정적 id)에 적합해요. 같은 명시적 run_id로 capability 인스턴스를 여러 .run() 호출에 재사용하면 before_run에서 ValueError가 발생해요. 도구 효과 원장이 (run_id, tool_call_id)로 키가 매겨지고 프로바이더가 결정적 도구 호출 id를 재사용하므로, 조용한 충돌이 unknown_after_crash 신호를 지우기 때문이에요. 다중 턴 그룹에는 conversation_id=를 사용하세요.
agent_name 설정, run_id 미설정 은 완전한 (agent_name, ctx.run_id) 쌍의 경로 안전 base64url 인코딩을 도출해요. 인코딩은 FileStepStore의 200자 한도 안에서 단사적이라, 재생이 승인된 서로 다른 컨텍스트 id 사이의 충돌 없이 같은 저장된 실행을 대상으로 해요. 더 긴 파생 id는 메모리·SQLite·Mongo 스토어를 포함해 백엔드 선택 전에 ValueError를 발생시켜요.
둘 다 미설정 은 ctx.run_id를 그대로 사용해요. 컨텍스트 run id가 없으면 RuntimeError가 발생해요. 하나를 만들면 재생된 쓰기가 분리되기 때문이에요.

지속 실행 (Durable execution)

StepPersistence는 안정적인 capability id step_persistence를 가지므로, id=를 넘기지 않고 Pydantic AI 지속성 capability와 함께 붙일 수 있어요. 같은 에이전트에 StepPersistence 인스턴스가 둘 이상일 때만 명시적 id를 넘기세요.

여섯 개 스토어 경계가 지속 작업이에요: 등록 신원, 실행 등록, 이벤트 추가, 스냅샷 저장, 도구 효과 시작, 도구 효과 완료/실패. 모든 영속 타임스탬프는 그 작업 중 하나 안에서 읽히므로, 재생은 workflow 벽시계를 다시 읽는 대신 저널링된 값을 사용해요. 저널링된 등록 신원은 재시도된 등록을 멱등하게 만들고, 같은 run id의 별개 재사용은 여전히 실패해요.

capability가 쓴 이벤트와 스냅샷은 결정적 실행별 멱등성 키를 지녀요. 모든 내장 스토어는 이미 적용한 키를 억제하고, idempotency_key=None으로 직접 만든 레코드는 추가 동작을 유지해요. 스냅샷 키는 step_index와 state와 함께 실행별 저장 시퀀스를 사용해요. 재생은 같은 시퀀스를 만들고, 같은 단계·상태의 서로 다른 스냅샷은 쓰기 순서와 더 새로운 히스토리를 유지해요.

오케스트레이터 패턴 — 하나의 논리 에이전트가 많은 턴을 서빙 — 은 공유 run_id가 아니라 conversation_id를 사용해요:

import asyncio

from pydantic_ai import Agent
from pydantic_ai_harness import StepPersistence
from pydantic_ai_harness.step_persistence import 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')

    # 이 오케스트레이터의 모든 턴, 시간순:
    records = await store.list_runs(conversation_id='orch-conv')


asyncio.run(main())

3단계 신원 (Three-level identity)

capability는 pydantic_ai의 신원 스택을 반영해요:

개념	정의	세분성
`conversation_id`	대화. pydantic_ai가 `Agent.run`의 `conversation_id=` 인자, `message_history`의 가장 최근 `conversation_id`, 또는 새 UUID7에서 해석.	실행의 시퀀스
`run_id`	하나의 `Agent.run` 호출.	시퀀스의 한 단계
`step_index`	실행 내 그래프 노드 수 (`ctx.run_step`).	한 실행 안의 한 노드

StepEvent.conversation_id와 RunRecord.conversation_id는 ctx.conversation_id에서 채워져요. 그래서 하나의 conversation_id를 공유하는 세 번의 .run() 호출은 세 개의 구분되는 run_id를 만들고, 모두 그룹으로 조회 가능해요:

import asyncio


async def main():
    runs = await store.list_runs(conversation_id='conv-abc')  # 3개 레코드, 시간순


asyncio.run(main())

위임의 조사 계속하기 (Continuing a delegate's investigation)

pydantic_ai에는 이미 "이전 컨텍스트로 계속"을 위한 message_history=가 있어요. StepPersistence는 병렬 메커니즘을 도입하지 않아요. 가장 최근 정착 스냅샷을 로드하는 헬퍼 하나를 노출해요:

import asyncio

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

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


async def main():
    # 앞서: 후속이 찾을 수 있도록 첫 턴에 conversation id를 붙인다.
    await librarian.run(
        'Find ThinkingPartDelta and confirm the callable allowance',
        conversation_id='libr-conv',
    )

    # 나중에 (아마 다른 프로세스에서):
    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',   # 대화 그룹 유지
    )


asyncio.run(main())

fork_run(store, run_id=...)는 같은 모양을 반환하지만, 호출자가 그 스냅샷 지점에서 분기된 논리 실행을 원할 때 사용해요(새 실행은 새 run_id를, 아마 새 conversation_id를 얻어요).

"계속해도 안전함"이 의미하는 것 (What "safe to continue from" means)

기본적으로 continue_run은 해당 run_id에 대한 최신 complete 스냅샷의 메시지를 반환해요 — 캡처 시점에 도구 작업이 완전히 정착된 지점이에요. 스냅샷은 다음 경계에서 쓰여요:

모든 도구 호출이 반환된 모든 CallToolsNode 후 — 보류 중인 도구 반환 요청이 접혀 들어가므로, 다음 모델 요청이 전송되기 전에 지점이 도구 완료 순간에 지속돼요.
실행이 그 경계를 지나 끝난 after_run에서(아무 경계에도 닿지 않은 실행, 또는 마지막 것 뒤에 닫는 응답이 떨어지는 Agent.run_stream).
실행이 _실패_할 때: 실패 시점의 살아있는 히스토리가 그 모양이 어떻든 저장돼요 — 깨끗한 도구 주기 후 raise하는 모델 요청은 complete 스냅샷을 만들고, 도구 주기 중간 크래시는 모든 완료된 주기를 지니는 interrupted 스냅샷을 만들어요.

interrupted 스냅샷은 재개 시 보낼 수 있어요 — pydantic-ai(>= 2.10)는 매 모델 요청 전에 깨진 도구 호출/결과 짝을 복구해요 — 하지만 반드시 안전한 것은 아니에요. 보류 중인 도구 호출이 (새 프롬프트 없이 재개하며) 재실행되거나 합성된 interrupted 반환으로 마감될 수 있는데, 어느 쪽도 원래 부작용이 일어났는지 말하지 않아요. 그것이 도구 효과 원장의 일이에요. 그래서 기본 읽기 경로는 interrupted 스냅샷을 건너뛰고, list_unresolved_tool_effects를 확인한 뒤 include_interrupted=True를 continue_run/fork_run/latest_snapshot에 전달하세요. 일치하는 스냅샷이 없으면 continue_run은 LookupError를 발생시켜요.

실행 계보: `parent_run_id`

parent_run_id는 기능적 의존성이 아니라 계보 라벨이에요. 두 가지를 해요:

모든 StepEvent와 RunRecord가 그것을 지니므로 필터링하고 그룹화할 수 있어요.
store.list_runs(parent_run_id='orch-1')는 그 오케스트레이터를 가리키는 모든 위임 실행을 반환해요.

프로세스 내 위임에서는 자동 추론돼요. 오케스트레이터의 도구가 위임의 Agent.run(...)을 동기적으로 호출하면, 위임의 StepPersistence가 오케스트레이터의 wrap_run이 설정한 ContextVar를 통해 오케스트레이터의 run_id를 집어요. 스레딩이 필요 없어요:

import asyncio

from pydantic_ai import Agent
from pydantic_ai_harness import StepPersistence
from pydantic_ai_harness.step_persistence import 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 자동 채움
    return result.output


async def main():
    # 아래 조회가 run_id를 찾을 수 있도록 오케스트레이터 턴에 태그.
    await orchestrator.run(
        'Where is ThinkingPartDelta defined?',
        conversation_id='orch-conv',
    )

    # 이제 모든 librarian 실행이 오케스트레이터의 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())

명시적으로 parent_run_id=를 설정해 재정의하세요(예: ContextVar가 전파되지 않는 프로세스 간 위임).

parent_run_id는 conversation_id와 구분돼요. 오케스트레이터와 위임은 보통 다른 대화에 살아요(오케스트레이터는 사용자와 이야기하고, 위임은 스스로와 이야기해요). 그러나 부모-자식 링크를 공유해요.

실행 트리 검사 (Inspecting a run tree)

list_runs는 모든 백엔드에 걸쳐 started_at 오름차순으로 정렬된 일치 항목을 반환해요. [-1]로 가장 최근 것을 고르세요.

import asyncio


async def main():
    # 한 오케스트레이터 실행의 모든 위임 (시간순)
    delegates = await store.list_runs(parent_run_id='orch-3f2a')

    # 한 대화의 모든 실행 (많은 .run() 호출을 가로지르는 다중 턴)
    turns = await store.list_runs(conversation_id='conv-abc')
    latest_turn = turns[-1]

    # 필터 결합 (AND):
    focused = await store.list_runs(
        parent_run_id='orch-3f2a',
        conversation_id='libr-conv',
    )

    # 실행별 상세:
    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():
    # 이전 위임 실행이 조사 중간에 죽었다.
    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'이고 종료 업데이트가 없음 -- 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}')

    # 재개할지 분기할지 결정:
    history = await continue_run(store, run_id='libr-3f2a')
    # 미해결 도구가 읽기 전용이고 다시 해도 안전하다면:
    await librarian.run('continue investigating', message_history=history,
                        conversation_id='libr-conv')
    # 부작용이 일어났을 수 있고 오케스트레이터가 새 시도를 원한다면:
    history = await fork_run(store, run_id='libr-3f2a')
    # ... 다른 agent_name / conversation_id로 새 위임 실행에 전달.

    # 위의 미해결 효과를 확인한 뒤, interrupted 프런티어 자체에서(크래시된 주기 포함) 재개하려면:
    history = await continue_run(store, run_id='libr-3f2a', include_interrupted=True)


asyncio.run(main())

부작용 중복 제거는 오케스트레이터의 책임이에요. 외부 상태를 쓰는 도구는 annotate_tool_effect로 진행 중인 ToolEffectRecord에 주석을 달아야 해요:

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)   # 실제 부작용
    return 'ok'

헬퍼는 StepPersistence ContextVar에서 활성 run_id를, ctx에서 tool_call_id/tool_name을 읽은 다음 이전 레코드에 메타데이터를 병합해요. step-persistence로 감싼 도구 호출 밖에서 호출하면 no-op이에요. after_tool_execute는 종료 completed/failed 항목을 쓸 때 두 필드를 모두 보존해요.

압축 영수증 핸들 (Compaction receipt handles)

StepPersistence.compaction_transcript_handle()은 현재 run_id를 압축 영수증에 노출해요. 이것은 이 스토어의 영속 실행 히스토리의 식별자이지, 압축 전 트랜스크립트가 계속 사용 가능하다는 약속이 아니에요. 스냅샷이 이미 압축된 히스토리를 담을 수 있고, 구성된 보존이 오래된 스냅샷을 삭제할 수 있어요.

백엔드 (Backends)

InMemoryStepStore — 프로세스 로컬, 테스트에 좋음.
FileStepStore(directory) — <directory>/<run_id>/ 아래 디렉터리 배치:
- run.json — RunRecord (계보)
- events.jsonl — 추가 전용 StepEvents
- tool_effects.jsonl — 추가 전용 ToolEffectRecords, 이 실행에 범위가 지정됨
- snapshot-keys.jsonl — 스냅샷 프루닝과 무관하게 유지되는 재생 억제 키
- snapshots/{seq}.json — ContinuableSnapshots, 실행별 단조 카운터로 이름 지정(step_index가 아니에요. 같은 run_id가 Agent.run 호출 간에 재사용될 때 충돌하기 때문이에요. ctx.run_step가 각 호출에서 0으로 리셋되거든요.)
SqliteStepStore(database='runs.db') — 테이블 runs, events, snapshots, snapshot_idempotency_keys, tool_effects와, 외부화된 blob용 형제 media 테이블을 가진 단일 SQLite 파일(미디어 지속 참고). WAL 모드가 활성화되고, tool_effects는 (run_id, tool_call_id)별 upsert라 최신 상태가 이겨요. 스냅샷은 FileStepStore._next_snapshot_seq를 반영하도록 AUTOINCREMENT seq를 사용해요. 스냅샷 state 열이 있기 전에 만든 데이터베이스는 열 때 자동으로 얻어요(기존 행은 complete로 읽혀요). 애플리케이션의 나머지와 sqlite3.Connection을 공유하려면 database= 대신 connection=을 전달하세요. 훅 호출이 워커 스레드에 발송되므로 연결은 check_same_thread=False로 열려야 해요.
MongoStepStore(client= or db_url=, database=...) — MongoDB 컬렉션 runs, events, snapshots, snapshot_idempotency_keys, tool_effects, counters(원자적 $inc가 단조 seq 할당). 실행 등록은 runs._id = run_id로 원자적 삽입을 사용하고, 중복 id는 ValueError를 발생시켜요. mongodb extra(이것이 pymongo>=4.17.0 설치)가 필요해요. 공유 AsyncMongoClient를 client=로, 또는 연결 문자열을 db_url=로 전달하세요(그러면 스토어가 클라이언트를 소유해요. 해제하려면 await store.aclose() 호출). media_threshold_bytes 이상의 개별 부분은 기본적으로 같은 클라이언트의 MongoMediaStore로 외부화돼요. 이것은 값을 단위로 한 오프로드이지, 총량 상한이 아니에요. 임계값 아래 많은 부분의 스냅샷은 여전히 MongoDB의 16 MiB 문서 한도를 초과해 삽입에 실패할 수 있으므로, 워크로드에 그 위험이 있다면 임계값을 낮추세요.

모두 같은 비동기 StepStore 프로토콜을 구현하므로, capability 훅은 파일/sqlite 백엔드에서 이벤트 루프를 절대 차단하지 않아요(anyio.to_thread로 I/O 발송). Mongo 백엔드는 네이티브 비동기예요.

FileStepStore는 경로 순회를 막기 위해 run_id를 [A-Za-z0-9_.-]{1,200}에 대해 검증해요(그리고 ..를 거부). 사용자 제어 id를 전달하는 호출자는 그래도 먼저 삭제해야 해요.

`MongoStepStore`가 첫 쓰기에 만드는 것

스토어는 첫 쓰기에서 열 개 인덱스의 createIndex를 발행해요: runs의 conversation_id와 parent_run_id(둘 다 sparse)와 started_at; events의 (run_id, seq)와 고유 키 (run_id, idempotency_key); snapshots의 (run_id, seq), 고유 키 (run_id, idempotency_key), (run_id, state, seq); tool_effects의 고유 (run_id, tool_call_id)와 (run_id, status). 멱등성 인덱스는 키가 문자열인 문서만 포함하므로 None은 추가 동작을 유지해요. 기본 MongoMediaStore는 미디어 페이지에 설명된 하나를 더 추가해요. 스토어를 기존 배포에 지정하기 전에 알 만한 세 가지 결과가 있어요:

연결 사용자는 인덱스를 만들 권한이 필요해요. 그것이 없는 제한된 Atlas 역할은 구성이 아니라 첫 쓰기에서 실패해요.
기존 tool_effects 컬렉션이 이미 중복 (run_id, tool_call_id) 쌍을 담고 있으면 고유 인덱스 빌드가 실패해요.
이미 채워진 컬렉션에 대한 인덱스 빌드는 그 첫 호출에서 시간과 I/O를 써요.

RunRecord.metadata와 StepEvent.metadata는 중첩 문서로 저장되므로, 그 키가 BSON 필드 이름이 돼요. .를 포함하거나 $로 시작하는 키는 MongoDB 5.0 이상이 필요하고, NULL 바이트를 포함하는 키는 서버에 닿기 전에 BSON 인코더가 거부해요. CI는 두 Mongo 백엔드를 mongo:8에 대해 실행해요.

MongoDB 지원 설치:

pip install "pydantic-ai-harness[mongodb]"

uv add "pydantic-ai-harness[mongodb]"

스냅샷 성장 경계 (Bounding snapshot growth)

각 단계는 증가하는 seq로 키가 매겨진 새 전체 히스토리 스냅샷을 쓰고, 기본적으로 아무것도 프루닝되지 않아요. 하나의 긴 Agent.run 안에서 스냅샷 수는 정착된 도구 호출 단계 수와 같으므로, 긴 단일 실행은 growing storage cost를 지불해요.

네 스토어 모두 — InMemoryStepStore, FileStepStore, SqliteStepStore, MongoStepStore — 선택적 max_snapshots_per_run: int | None(기본 None, 무제한 — 이전 동작과 byte-for-byte 동일)을 받아요. N >= 1로 설정하면 각 save_snapshot이 실행을 보존 집합으로 프루닝해요:

seq 기준 최신 N개 스냅샷,
전체 기준 최신 스냅샷(latest_snapshot(include_interrupted=True) 서빙),
최신 complete 스냅샷(기본 읽기 경로 서빙).

마지막 둘은 최신 N개 스냅샷이 모두 interrupted이고 최신 재개 가능 complete가 그 창 아래에 있어도 두 읽기 모드를 모두 올바르게 유지하므로, 보존 집합이 N을 초과할 수 있어요. from_spec(..., max_snapshots_per_run=N)은 그 경계를 그것이 만드는 스토어(backend='memory', 'file', 'sqlite'; Mongo 스토어는 spec이 아니라 직접 빌드)에 전달해요.

from pydantic_ai_harness.step_persistence import FileStepStore

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

스냅샷을 프루닝해도 그 외부화된 미디어는 삭제되지 않아요. blob은 콘텐츠 주소가 있고 스냅샷·실행 간에 공유될 수 있으므로, 고아 blob GC는 범위 밖이에요(비목표 참고). 연령 기반(TTL) 만료도 범위 밖이에요. 스냅샷별이 아니라 전체 실행 세분성에 속하기 때문이에요.

경계 보존은 압축 전 스냅샷을 포함해 오래된 단계별 스냅샷을 버려요. 실행의 보존된 스냅샷을 합집합하여 히스토리를 재구성하는 다운스트림 — 스냅샷 검색이나 run_id로 키가 매겨진 압축 영수증 — 은 보존된 것만 볼 수 있어요. 빡빡한 경계(예: max_snapshots_per_run=1)에서는 오래된 압축 전 상태가 사라지므로, 그 경계를 그러한 복구가 닿을 수 있는 한계에 대한 하드 상한으로 취급하세요. 역사적 재구성이 중요하면 경계를 None으로 두거나 회복해야 할 히스토리를 덮을 만큼 높게 설정하세요.

미디어 지속 (Persisting media)

스냅샷 안에 base64로 인라인된 BinaryContent 페이로드(이미지, 오디오, 문서, 비디오)는 메시지를 담은 모든 파일·행을 부풀려요. 큰 텍스트 부분(예: 큰 도구 반환 문자열)도 마찬가지이며 MongoStepStore 스냅샷을 MongoDB의 16 MiB 문서 상한(#440)을 넘게 밀 수 있어요. 파일·sqlite·mongo 백엔드는 구성된 MediaStore를 통해 어떤 BinaryContent.data와 문자열 content가 64 KiB 이상인 부분을 외부화하고, 스냅샷에 URI 참조를 남겨요. 같은 media_threshold_bytes가 바이너리와 텍스트 모두를 지배하며, 별도 텍스트 노브는 없어요. 왕복은 투명해요. latest_snapshot(...).messages[*]는 원래 BinaryContent 바이트와 텍스트를 반환해요.

텍스트 외부화는 Mongo 전용이 아니며 media_store=None 말고는 탈출구가 없어요. 워커가 공유되므로 기존 FileStepStore/SqliteStepStore 배포는 이 릴리스부터 큰 텍스트 부분과 바이너리 부분 모두에 blob을 쓰기 시작해요. 그 전에 쓴 스냅샷은 여전히 복원돼요. 판독기가 오래된 바이너리 마커 모양을 인식하기 때문이에요. 이 호환성은 업그레이드 전용이에요. 텍스트 외부화보다 오래된 릴리스는 모든 마커를 바이너리로 취급하므로, 외부화된 텍스트 마커를 담은 스냅샷을 검증할 수 없어요. 그 마커를 담은 영속 스냅샷에는 현재 판독기를 유지하세요.

예약 키 이스케이핑은 이 스토어들에 같은 규칙을 가진 두 번째 마커 형식 세대예요. 마커 형식의 이름공간 키를 사용하는 페이로드는 버전이 매겨진 예약 매핑(__harness_external_escaped_keys__ 스태시, __harness_external_marker_format__ 아래 형식 버전으로 스탬프)으로 이동되고, 현재 판독기는 그 값을 자체 키로 되돌려요. 반대 방향 호환성은 업그레이드 전용이에요. 이스케이핑 형식보다 오래된 판독기는 외부화된 필드를 올바르게 다시 인라인하지만, 값이 아니라 두 예약 키를 복원된 페이로드에 남겨둬요. 이 판독기가 모르는 버전으로 스탬프된, 둘 다 담은 마커는 예약 값이 제거된 채 복원되는 대신 거부돼요. restore_media는 ValueError를 발생시키고, latest_snapshot은 파일·sqlite·mongo 스토어에 대해 호출자에게 표면화해요. list_snapshots는 달라요. 각 스토어는 실패한 스냅샷을 구문 분석 불가로 취급하고 건너뛰며 오류를 기록하므로, 알 수 없는 버전은 예외가 아니라 누락된 스냅샷으로 나타나요. 그 거부가 버전 게이트이며 의도된 것이지만, 스토어 사용자가 대비해야 해요. 이스케이프된 마커를 담은 영속 스냅샷에는 현재 판독기를 유지하세요.

StepStore	기본 `media_store`	blob이 사는 곳
`InMemoryStepStore`	해당 없음	바이트는 인메모리 스냅샷에 유지
`FileStepStore`	`DiskMediaStore(<root>/media/)`	`<root>/media/<sha256>.bin`
`SqliteStepStore`	`SqliteMediaStore(database=<same db>)`	같은 DB의 형제 `media` 테이블
`MongoStepStore`	`MongoMediaStore(client=<same client>)`	형제 `media` + `media_chunks` 컬렉션

자신의 MediaStore를 전달해 대상을 재정의하세요:

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

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,  # 원하면 올리거나 내리세요
)

완전히 탈출하세요(바이트를 스냅샷 JSON/행에 인라인으로 유지):

from pydantic_ai_harness.step_persistence import FileStepStore, SqliteStepStore

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

URI는 media+sha256://<hex>로, 콘텐츠 주소가 지정돼요. 어떤 MediaStore를 통해 쓰여도 같은 blob은 같은 방식으로 해석되므로 중복 제거가 자동이고, 기반 저장소를 옮기는 것은 한 줄 교체예요. 제공되는 구현은:

DiskMediaStore(directory) — <directory>/<sha256>.bin에 blob당 파일 하나.
SqliteMediaStore(database=...) 또는 SqliteMediaStore(connection=...) — blob당 행 하나(INSERT OR IGNORE로 콘텐츠 주소 중복 제거).
S3MediaStore(bucket=, endpoint=, region=, access_key_id=, secret_access_key=) — 경로 스타일 URL + 수제 SigV4. AWS S3, Cloudflare R2(region='auto'), MinIO, 기타 S3 호환 프로바이더와 호환. PUT/GET/HEAD만 — v1에는 multipart, 라이프사이클, 리스팅 없음.
MongoMediaStore(client= or db_url=, database=...) — MongoDB, mongodb extra 필요. 각 blob은 media 매니페스트 문서와 형제 media_chunks 컬렉션을 가로지르는 sha256 주소 청크(GridFS가 아닌 수동 청킹이라 중복 제거가 보존돼요. 미디어 페이지 참고)이므로, BSON 문서 하나보다 큰 blob도 저장하고 읽어 오나요. 청킹은 문서를 경계로 하지 메모리를 경계로 하지 않아요. 스트리밍 API가 없어서 각 blob은 put과 get 모두에서 프로세스 메모리에 통째로 담겨요. 매니페스트는 MediaContext.metadata를 인라인으로 담고 청크되지 않으므로, blob별 메타데이터를 작게 유지하세요. collection=은 두 컬렉션을 모두 이름 변경하고 chunk_size_bytes=(기본 8 MiB)는 분할 크기를 설정해요.

외부화된 바이트를 URL로 노출 (Exposing externalized bytes as URLs)

각 스토어는 정규 media+sha256://<hex> URI를 모델이 직접 가져올 수 있는 URL로 바꾸는 public_url= callable을 받아요. 곧 나올 MediaExternalizer capability는 이것을 사용해 모델이 메시지를 보기 전에 BinaryContent 부분을 ImageUrl/AudioUrl/기타 URL 부분으로 바꿔, 프로바이더가 바이트를 요청 body로 재인코딩하지 않고 큰 미디어를 와이어 너머로 가져오게 해요.

정적 기본 URL(공개 R2 버킷, 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/'),
)

사전 서명 또는 회전 서명 URL — (uri, MediaContext)를 받는 async callable을 전달하세요:

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`, 확장 가능한 작업별 가방

모든 MediaStore 메서드(put, get, exists, public_url, get_metadata)와 두 사용자 제공 callable(PublicUrlResolver, KeyStrategy)은 MediaContext를 받아요:

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


@dataclass(frozen=True, kw_only=True)
class MediaContext:
    media_type: str | None = None                    # 예: 'image/png'
    filename: str | None = None                      # 원래 파일 이름, 알 때
    metadata: Mapping[str, str] = field(default_factory=dict)  # 사용자 제공 태그

모든 필드는 기본값이 있고, 사용 사례가 생기면 새 필드가 비파괴적으로 추가돼요. 가진 것을 전달하고, 나머지는 무시하세요.

스토어별 지속. get_metadata(uri)는 네 스토어 모두에서 사용자 제공 metadata 매핑을 왕복시켜요. media_type도 영속되지만 get_metadata가 반환하는 것의 일부는 아니에요(바이트 페이로드 자체를 위해 저장되며, 예를 들어 Content-Type으로 사용).

SqliteMediaStore는 metadata를 JSON 열에, media_type을 전용 열에 써요.
S3MediaStore는 metadata를 서명된 x-amz-meta-* 헤더(ASCII 영숫자 + 대시 키 이름)로, media_type을 Content-Type으로 보내요. get_metadata는 HEAD 응답에서 x-amz-meta-* 값을 다시 읽어요.
DiskMediaStore는 각 blob 옆에 사이드카 JSON 파일(<resolved>.meta.json)을 tmp + rename으로 원자적으로 써요. 사이드카는 put이 메타데이터를 지니지 않았을 때만 없어요.
MongoMediaStore는 metadata를 JSON 문자열로, media_type을 blob 매니페스트 문서(기본 media 컬렉션)의 전용 필드로 써요. get_metadata는 JSON 문자열을 다시 디코딩해요. 매핑이 중첩 필드가 아니라 JSON 문자열 하나이므로, 메타데이터 키는 여기서 BSON 필드 이름 규칙의 대상이 아니에요.

`key_strategy`: 백엔드 저장 경로 제어

기본은 <sha256>.bin이에요. DiskMediaStore와 S3MediaStore는 기존 배치에 맞춰 재정의를 받아요. SqliteMediaStore와 MongoMediaStore는 받지 않아요(다이제스트가 기본 키라 사용자 선택 키는 중복 제거를 깨거나 no-op이 될 것이므로, 행·문서를 옮기려면 table=/collection= 사용):

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)

주의: 전략이 context.media_type에 의존하면(예: 확장자 선택), 읽기 시 같은 컨텍스트가 주어지지 않으면 get(uri)와 exists(uri)가 blob을 찾지 못해요. 순수 경로 조직 전략(컨텍스트 의존 없음)에는 그 제약이 적용되지 않아요.

DiskMediaStore는 스토어 디렉터리를 벗어나는 것을 막기 위해 절대 경로나 .. 세그먼트를 만드는 전략을 거부해요.

별도로, 네 스토어 모두 public_url= 해석기를 받아요. CDN, 로컬 HTTP 서버, 서명 URL 서비스가 바이트 앞에 설 때 유용해요. 그것이 없으면 public_url(...)는 None을 반환해요(해석기가 구성되고 문자열을 반환할 때만 모델이 URL을 봄).

pydantic_ai 프로바이더는 대상 모델이 그 URL 유형을 네이티브로 받지 않을 때 URL에서 바이트를 투명하게 다운로드하므로, URL을 내는 것은 항상 안전해요. 와이어 절감만 잃을 뿐, 정확성은 절대 잃지 않아요.

곧 나올 MediaExternalizer capability — 그것이 도착하면 조합은 Agent(capabilities=[MediaExternalizer(store), StepPersistence(...)])가 되고 StepPersistence는 이미 URL화된 메시지를 보게 되어 외부화 워커가 no-op이 돼요. 기존 API는 바뀌지 않아요.

지원되지 않는 백엔드로 지속 (Persisting to unsupported backends)

DynamoDB, Postgres, Redis, GCS, 기타 백엔드는 이 릴리스 범위 밖이에요. 자신의 StepStore(Protocol의 메서드 약 10개)나 자신의 MediaStore(다섯 메서드: put, get, exists, public_url, get_metadata)를 쓰고 store=/media_store=로 전달하세요. 하나 배송했다면 이슈를 열어 주세요. 추상화하기 전에 N >= 3 실제 구현으로 결국 공유될 어댑터 계층을 채우고 싶어요.

대화 헤드와 이름 붙이기 (Conversation heads and background names)

pydantic_ai_harness.step_persistence.conversations는 다중 턴 애플리케이션을 위한 SqliteConversationStore, ConversationSummary, SavedConversation를 제공해요. 대화 헤드는 실행별 체크포인트와 구분돼요. 허용된 프롬프트와 압축 같은 실행 간 편집을 포함해요. 겹치는 실행 스냅샷을 연결해 재구성하지 마세요.

save(summary=..., messages=...)는 제공된 콘텐츠 리비전을 비교하고 커밋된 요약을 반환해요. 오래된 작성자나 삭제된 세션은 ConversationConflict를 발생시켜요. get(conversation_id=...)는 단계 스냅샷이 쓰는 것과 같은 미디어 형식을 통해 메시지를 복원해요. listing(query=..., limit=..., offset=...)는 메시지를 로드하지 않고 요약을 반환해요. 검색은 유니코드 케이스 폴딩으로 저장된 사용자/어시스턴트 텍스트와 메타데이터를 일치시켜, 다중 모드 프롬프트 안의 텍스트 항목을 포함해요. 검색은 도구 출력, reasoning, 버려진 압축 전 히스토리를 포함하지 않아요. 알 수 없는 메타데이터 스키마 버전은 거부돼요.

메타데이터 이름 붙이기는 별도 버전을 사용해요. name(source=..., title=..., ...)는 더 새로운 콘텐츠 리비전, 더 새로운 이름, 또는 수동 제목을 덮어쓸 수 없어요. 이름 붙이기는 활동 타임스탬프를 바꾸지 않아요. delete(source=...)는 공유 미디어를 유지하면서 같은 SQLite 데이터베이스에서 대화와 연관된 실행 기록을 원자적으로 제거해요. 안전한 삭제가 아니에요. 로컬 라이브 PID가 미완료 대화를 바쁜 것으로 표시해요. 이것은 분산 임대가 아니며 데이터베이스를 호스트 간에 공유해서는 안 돼요. PID 재사용은 보수적으로 바쁜 것으로 취급돼요.

데이터베이스는 지원되는 곳에서 소유자 전용으로 만들어져요. 콘텐츠는 암호화되지 않아요. 자동 대화 TTL이나 미디어 가비지 컬렉션은 없어요.

pydantic_ai_harness.step_persistence.naming은 도구 없는 이름 붙이기 에이전트와 SessionNamer(애플리케이션 작업 그룹이 소유하는 워커)를 제공해요. submit(id)는 열 개 세션으로 경계를 둔 큐에서 작업을 병합해요. run()은 소유자가 취소할 때까지 한 번에 하나의 작업을 처리해요. backfill(entries)는 최신 열 개 항목을 고려해요. 이름 붙이기 실패는 debug 레벨로 기록되고 기존 메타데이터를 사용 가능하게 남겨요. 취소는 전파돼요. 애플리케이션은 워커의 모델 클라이언트나 저장소 의존성을 닫기 전에 워커에 조인해야 해요.

이름은 짧은 제목, 부제, 최대 네 개 태그로 구성돼요. 모델은 이전 제목/상세와 경계를 둔 2,400자 현재 대화 꼬리를 받아요. 이것은 의도적으로 메시지 인덱스 커서가 아니에요. 압축과 복구가 목록을 대체할 수 있기 때문이에요. 생성된 이름은 16개 콘텐츠 리비전 후 다시 자격을 얻어요. 수동 이름은 바뀌지 않아요. 이름 붙이기 요청은 60초 워커 타임아웃이 있고, 제공된 generate_name 헬퍼는 최대 두 모델 요청과 250 출력 토큰을 허용해요. 헬퍼의 에이전트는 session_namer라 불리고 도구가 없으며 포그라운드 에이전트의 capability를 상속하지 않아요.

Core의 에이전트 스팬은 보조 모델 호출을 session_namer에 귀속시켜요. 두 번째 스팬 계층은 방출되지 않아요. 성공적인 이름 붙이기 응답 토큰 수는 포그라운드 히스토리와 별도로 저장돼요. 세션이 여전히 존재할 때 오래된 것으로 거부된 결과를 포함해요. 실패하거나 타임아웃된 요청은 애플리케이션에 사용할 수 없는 프로바이더 사용량을 발생시킬 수 있어요. 보조 호출의 금전적 가격은 유지된 히스토리 비용에 포함되지 않아요. 애플리케이션은 이름 붙이기 모델을 선택하고 추가 프로바이더 요청을 사용자에게 공개해요.

이른 체크포인트와 알림 (Earlier checkpoints and notifications)

capture_frontier=True를 설정하면 모델 요청 전에 허용된 요청 히스토리와 도구 실행 전에 모델 응답 프런티어를 저장해요. 기본은 False로 기존 체크포인트 빈도를 보존해요. CLAI가 그것을 켜요. 그러면 첫 모델 요청 실패가 프롬프트를 유지하고, 도구 주기 중간에 죽은 프로세스가 주기가 정착되기 전에도 제안된 호출과 인자를 유지할 수 있어요.

pydantic_ai_harness.step_persistence.recovery의 inspect_recovery(store=..., run_id=...)는 최신 및 정착 스냅샷, 미해결 효과, 기록된 완료/실패 도구의 이름을 반환해요. 효과가 재생에 안전하다고 추론하지 않아요.

이것들은 여전히 메시지 체크포인트이지 그래프 상태 체크포인트가 아니에요. 미정착 프런티어의 스냅샷은 interrupted이며 기본 읽기 경로에서 벗어나 있어요. after_run은 같은 길이·단축된 히스토리 재작성을 잡기 위해 메시지 수뿐 아니라 최종 콘텐츠를 비교해요. 기록자를 after_run이 히스토리를 변형하는 capability 앞에 두세요. core는 after 훅을 역순으로 실행하거든요. 스냅샷 메시지 값은 저장 전에 복사돼 공유 참조로 나중 변형이 저장된 인메모리 체크포인트를 바꿀 수 없게 해요.

SnapshotSaved는 체크포인트 쓰기가 끝난 뒤 방출되는 타입 지정 capability 이벤트예요. persistence_run_id, conversation_id, step_index, state를 지녀요. core의 hooks.on.event(SnapshotSaved)나 CLAI의 host.on(SnapshotSaved)로 구독하세요. 스토어 쓰기가 진실의 원천이에요. 알림은 지속 재생 중 반복될 수 있고, 관찰자 실패는 커밋된 쓰기를 되돌릴 수 없어요.

더 강한 interrupted 단계 복구를 위한 core 경계

자동 실행 복구는 구현되지 않았어요. 그것을 약속하기 전에 두 core 계약이 다뤄져야 해요:

on_run_error는 권위 있는 정리 후 히스토리를 노출해야 해요. 오늘 Harness는 노드/요청 훅에서 살아있는 목록 참조를 스태시해요. 바깥 오류 컨텍스트가 시작 시 목록을 참조할 수 있기 때문이에요. 그것은 core가 작업 히스토리를 제자리에서 계속 변형하는 것에 달려 있어요. core의 취소 결과 API는 호출자에게 유용하지만, 모든 오류 훅에 대해 같은 계약을 세우지 않아요.
기다리는 체크포인트 경계는 각 도구가 정착할 때 정규화된 결과를 노출해야 해요. 동반 사용자 콘텐츠, 재시도, 병렬 형제를 포함해서요. after_tool_execute는 모든 정규화 전에 원시 결과를 보고, after_node_run은 정착된 배치를 봐요. FunctionToolResultEvent는 정규화된 결과를 노출하지만, 스트림을 관찰하는 것은 그 결과를 도구 효과 원장 및 실행 프런티어와 원자적으로 커밋하는 것이 아니에요.

그래서 병렬 배치 중 하드 킬은 결과가 영속되지 않은 완료된 효과를 남길 수 있어요. started 효과는 크래시 후 알 수 없고, failed 도구조차 부분 외부 변경을 했을 수 있어요. 더 오래된 complete 체크포인트로 돌아가도 그 변경을 되돌리지 않아요. 외부 효과가 있는 도구는 자체 멱등성/조정 전략이 필요해요. Harness 이벤트는 외부 부작용을 로컬 SQLite 쓰기와 원자적으로 만들 수 없어요.

테스트는 실제 서브프로세스 킬, 조기 요청 실패, 최종 히스토리 재작성, 리비전 충돌, 경계/취소된 이름 붙이기를 다뤄요. 킬 테스트는 프런티어 캡처가 오류 훅 없이도 생존함을 확인해요. 정확히 한 번 실행 보장은 아니에요.

이 capability가 하지 않는 것 (What this capability does not do)

실행별 capability 상태, 그래프 노드 상태, 재시도 카운터, 진행 중 스트리밍 응답을 복원하지 않아요.
재생된 부작용을 자동으로 중복 제거하지 않아요. 아티팩트, 라벨, PR, 외부 상태를 쓰는 도구는 annotate_tool_effect(store, ctx, ...)를 호출해 오케스트레이터가 재생이 안전한지 결정하게 해야 해요(실패 복구 참고).
이벤트를 프루닝하지 않고, 기본적으로 스냅샷도 프루닝하지 않아요. 보존은 호출자의 책임이에요. 스냅샷 성장은 max_snapshots_per_run으로 선택적으로 경계할 수 있어요(스냅샷 성장 경계 참고).
외부화된 미디어를 가비지 컬렉션하지 않아요. 스냅샷을 프루닝해도 콘텐츠 주소 blob은 그대로 있어요. 스냅샷·실행 간에 공유될 수 있기 때문이에요.
OpenTelemetry 스팬을 방출하지 않아요. pydantic_ai의 Instrumentation capability가 이미 agent run/chat/running tool을 스팬하고 baggage로 gen_ai.agent.name, gen_ai.agent.call.id, gen_ai.conversation.id를 채워요. 향후 변경이 활성 스팬에 단계 지속성 속성을 추가할 수 있으며, 그것은 후속 이슈로 추적돼요.

API 참조 (API reference)

StepPersistence

Base: AbstractCapability[AgentDepsT]

추가 전용 단계 로그 + 재개 가능 스냅샷 + 도구 효과 원장. capability는 모든 흥미로운 경계(실행/모델 요청/도구 호출 시작, 완료, 실패)에서 StepEvent를 방출하고, 도구 호출마다 ToolEffectRecord를 기록해 오케스트레이터가 재생이 안전한지 결정하게 하며, 정착된 CallToolsNode 경계마다 ContinuableSnapshot을 저장해요. 보류 중인 도구 반환 요청을 접어 넣어 그 지점이 도구가 완료되는 순간 지속되게 하고, 실행이 그 경계를 지나 끝나면 after_run에서 폴백 저장도 해요. _실패_한 실행은 도구 작업 상태로 분류된 실패 시점의 살아있는 히스토리를 저장해요. 모든 도구 호출이 해결되면 complete, 그렇지 않으면 interrupted.

before_tool_execute와 after_tool_execute 사이에 크래시한 실행은 보이는 이벤트 흔적, started 도구 효과 기록(unknown_after_crash 신호), 모든 완료된 주기를 지니는 interrupted 스냅샷을 남겨요. 기본 latest_snapshot/continue_run 읽기 경로는 complete 스냅샷만 반환해요. list_unresolved_tool_effects를 확인한 뒤 include_interrupted=True를 넘겨 interrupted 프런티어에서 재개하세요.

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')

continue_run(store, run_id=...)/fork_run(store, run_id=...)로 이전 스냅샷을 로드한 뒤 그 결과를 Agent.run(..., message_history=...)에 전달하세요.

속성 (Attributes)

store — 이벤트, 스냅샷, 도구 효과를 기록하는 백엔드. 타입: StepStore 기본: field(default_factory=InMemoryStepStore)
agent_name — 논리 에이전트 이름(예: code_librarian, reproducer). 컨텍스트 파생 run_id의 안정적 접두사로 사용되어 스토어 검사가 에이전트와 지속 실행을 모두 식별하게 해요. 타입: str | None 기본: None
run_id — 이 한 번의 Agent.run 호출 식별자. run_id는 호출별이며 pydantic_ai.RunContext.run_id와 일치해요. 다중 턴 논리 그룹에는 Agent.run(...)의 conversation_id를 사용하세요. 그것이 pyai 네이티브 원시형이에요. 해석 순서(for_run에서 구체화): 1) 명시적 값 → 그대로 사용. 단일 샷 사용 사례: 테스트·재생·디버깅용 결정적 id. 같은 명시적 run_id로 capability를 여러 .run() 호출에 재사용하면 before_run에서 ValueError 발생 — 도구 효과 원장이 (run_id, tool_call_id)로 키가 매겨지고 프로바이더가 결정적 도구 호출 id를 재사용하므로, 조용한 충돌이 unknown_after_crash 신호를 지우기 때문. 다중 턴 그룹에는 Agent.run의 conversation_id= 사용. 2) agent_name 설정, run_id 미설정 → 두 값의 경로 안전 인코딩. 3) 둘 다 미설정 → .run()별 ctx.run_id. 타입: str | None 기본: None
parent_run_id — 이 실행을 만들어낸 실행. 둘러싸는 StepPersistence wrap_run 범위에서 자동 추론 — 오케스트레이터의 도구가 위임의 Agent.run(...)을 동기적으로 호출하면 위임이 여기서 오케스트레이터의 run_id를 수동 스레딩 없이 집어요. 명시적으로 설정해 재정의하세요(예: ContextVar가 전파되지 않는 프로세스 간 위임). 타입: str | None 기본: None
metadata — RunRecord와 각 이벤트에 저장되는 자유 형식 메타데이터. 타입: dict[str, str] 기본: field(default_factory=_empty_metadata)
capture_frontier — 실행 전에 허용된 입력과 모델 도구 호출 프런티어도 체크포인트. 이 추가 쓰기는 첫 요청 실패와 정착 도구 주기 전 프로세스 킬을 검사 가능하게 해요. 부작용을 재생 안전하게 만들지 않아요. 타입: bool 기본: False

메서드 (Methods)

from_spec (@classmethod) — 직렬화된 spec에서 구성. backend='memory'(기본), backend='file'(directory 포함), backend='sqlite'(database 포함)을 지원해요. 다른 backend 값은 ValueError 발생 — 인메모리 저장으로 조용히 폴백하면 오타가 우연한 비지속성이 되기 때문. max_snapshots_per_run(기본 None, 무제한)은 구성된 스토어에 전달돼 실행별 스냅샷 성장을 경계해요.
compaction_transcript_handle — 압축 영수증용 이 실행 트랜스크립트의 검색 핸들. TranscriptHandleProvider 프로토콜을 구조적으로 충족해요(import 결합 없음). 압축 전략은 RunContext.capabilities를 통해 이 capability를 발견하고 run id를 기록해요. for_run이 id를 구체화하기 전에는 None을 반환해요.
for_run (@async) — 이 Agent.run 호출에 대한 run_id와 parent_run_id를 구체화. 로컬 실행이 덮어쓰기 전에 둘러싸는 StepPersistence.wrap_run이 설정한 contextvar를 읽으므로, 위임의 parent_run_id가 오케스트레이터의 run_id를 가리키게 돼요. pydantic_ai 자신의 교차 실행 신호(RUN_ID_BAGGAGE_KEY via OTel baggage, RunContext.run_id, _CURRENT_RUN_CONTEXT)는 단일 슬롯이라 중첩 capability가 부모를 보기 전에 내부 Instrumentation.wrap_run이 덮어쓰기 때문에 별도 ContextVar가 필요해요. 하니스 로컬 contextvar는 로컬 wrap_run이 재바인딩하기 전에 여기서 부모를 스냅샷하게 해요.
wrap_run (@async) — 이 실행의 id를 contextvar에 푸시해 중첩 위임이 읽게 해요.
before_run (@async) — 실행 계보를 등록하고 run_started를 방출. 별개 등록에 의한 재사용을 거부 — 도구 효과 원장이 (run_id, tool_call_id)로 키가 매겨지고 프로바이더가 결정적 도구 호출 id를 재사용하므로, 같은 run_id의 두 번째 Agent.run은 조용히 충돌할 것. 저널링된 등록 신원이 이 지속 작업의 재시도를 새 실행과 구분해요.
after_run (@async) — run_completed를 방출하고 폴백으로만 최종 스냅샷 저장. 종료 CallToolsNode가 after_node_run에서 이미 최종 히스토리를 저장했다면 올바른 step_index를 지니지만, after_run 시점에는 ctx.run_step가 0으로 리셋돼 있어 재저장은 꼬리를 복제하고 오해의 소지가 있는 step_index를 찍을 것. 최종 콘텐츠가 최신 경계 스냅샷과 다를 때만 저장해요. 같은 길이 재작성 포함. 비교는 외부 스토어 읽기(지속 재생 중 바뀔 수 있음)가 아니라 실행별 복사 상태를 사용해요. 이것은 전혀 프로바이더 유효 경계에 닿지 않은 실행과, 종료 CallToolsNode가 아니라 SetFinalResult로 끝나고 마지막 경계 뒤에 닫는 응답을 추가하는 Agent.run_stream을 다뤄요.
on_run_error (@async) — 실행의 마지막 재개 지점으로 실패 시점의 살아있는 히스토리를 영속한 뒤 run_failed를 방출. 단일 오류 경로 저장 지점이에요. after_node_run이 스태시한 목록 참조를 읽는데, 그 시점 그 콘텐츠는 실행이 실패할 때까지 만든 전체 히스토리예요. 실패한 모델 요청 페이로드와 언와인드 중 그래프가 잡은 부분 도구 반환 포함. 스토어에 대해 아무것도 비교하지 않아요. 살아있는 히스토리는 정의상 최신 상태이므로 이전 경계 스냅샷은 단순히 대체되고, 스티키 프로세서가 다듬은 히스토리는 다듬어진 대로 영속돼요. 다음 요청이 보냈을 것과 정확히 같아요. 히스토리가 모델 응답을 담을 때마다 저장되며(베어 프롬프트는 실행 재시작과 같음), 모든 도구 호출이 해결되면 complete, 그렇지 않으면 interrupted로 분류돼요. interrupted 스냅샷은 기본 latest_snapshot 읽기 경로 밖에 있어요.
on_model_request_error (@async) — model_request_failed를 방출하고 재-raise. 여기서 스냅샷을 저장하지 않아요. 실패 요청 페이로드가 이미 살아있는 히스토리에 있고(그래프가 보내기 전에 요청을 추가), on_run_error의 저장이 그것을 덮기 때문이에요. 모델 계층이 회복하는 실패(재시도, 폴백)는 아무 구조가 필요 없어요.
after_node_run (@async) — 정착된 CallToolsNode 후 재개 가능 스냅샷을 저장하고 살아있는 히스토리 스태시를 갱신. 그 경계에서 앞선 ModelRequestNode의 모든 도구 호출에 일치하는 도구 반환이 있으므로 히스토리가 프로바이더 유효해요. 반환된 ModelRequestNode는 그 반환을 지니고 아직 ctx.messages에 없으므로, 검증 전에 그 요청이 접혀 들어가요. 그것 없이는 완료된 도구 호출 직후 죽은 워커가 재개 지점을 전혀 남기지 못해요(#373). is_provider_valid는 커스텀 노드가 히스토리를 재구성하는 경우의 방어 역할을 해요. after_run은 저장된 콘텐츠를 다른 capability의 같은 길이 재작성 포함해 최종 히스토리와 비교해요. 이 저장이 지속적인 것이에요. 실행이 여전히 건강한 동안 스토어에 들어가므로 훅을 발화하지 않는 하드 킬을 견뎌요. 오류 경로(on_run_error)는 raise가 언와인드하는 히스토리만 구조해요. 모든 노드 경계도 살아있는 메시지 목록을 다시 스태시해, 이후 노드가 자체 after_node_run 전에 raise하면 on_run_error가 실패 시 히스토리를 영속하게 해요. 스태시는 그 목록을 참조로 담으므로, 스냅샷 후보는 더하는 게 아니라 새 목록에 재바인딩돼요. 더하면 result.request가 오류 경로가 나중에 읽는 히스토리로 새어, 그래프가 요청 자체를 추가할 때 중복돼요.

Step Persistence