Python: [Feature]: Make workflows stateless with caller-owned checkpoints and msgspec snapshots

Start with the Workflow, WorkflowBuilder, runner, AgentExecutor, Checkpoint, and CheckpointStorage entry points described in the issue. Trace where mutable execution state, executor sessions, edge state, and snapshots are currently retained. Done means caller-owned checkpoints support stateless concurrent workflow runs, restartable snapshot branches, atomic persistence, and the proposed v2 behavior.

Description

Summary

For v2, align workflow state ownership with the agent/session model without introducing a separate WorkflowSession abstraction:

• agent.create_session() creates caller-owned state.
• agent.run(..., session=session) uses and updates that state.
• workflow.create_checkpoint() should create caller-owned workflow state.
• workflow.run(..., checkpoint=checkpoint) should use and update that state.

A Checkpoint represents one logical workflow execution and its lifecycle. It contains snapshots, and each snapshot contains the complete state required to resume from that superstep.

The developer owns the checkpoint and its lifecycle. The Workflow becomes a stateless, reusable definition.

This is a breaking change and should only be implemented for v2, hence the vnext label.

Current behavior

Today, mutable execution state is retained by Workflow, its runner, runner context, executors, and edge runners. Checkpoint storage can be attached to WorkflowBuilder or supplied to Workflow.run():

storage = FileCheckpointStorage("./checkpoints")

builder = (
    WorkflowBuilder(
        start_executor=start,
        checkpoint_storage=storage,
    )
    .add_edge(start, worker)
    .add_edge(worker, worker)
)

workflow = builder.build()
await workflow.run(message=10)

latest = await storage.get_latest(workflow_name=workflow.name)
if latest is None:
    raise RuntimeError("No checkpoint was created")

resumed_workflow = builder.build()
await resumed_workflow.run(checkpoint_id=latest.checkpoint_id)

This has several consequences:

• A Workflow instance owns mutable execution state.
• One workflow instance cannot safely execute independent runs concurrently.
• Checkpoint lifecycle is split between the workflow, storage, and application.
• Applications must query storage to discover the latest checkpoint.
• get_latest(workflow_name=...) is ambiguous when multiple executions use the same workflow definition.
• Resume is expressed through checkpoint_id and checkpoint_storage arguments rather than a caller-owned state object.
• FileCheckpointStorage uses JSON containing pickle/base64 payloads, while agent sessions use typed msgspec serialization.

Proposed developer experience

The workflow creates a checkpoint, and the caller passes that checkpoint to related runs:

storage = FileCheckpointStorage("./checkpoints")

workflow = (
    WorkflowBuilder(start_executor=start)
    .add_edge(start, worker)
    .add_edge(worker, worker)
    .build()
)

checkpoint = workflow.create_checkpoint(
    checkpoint_id="factor-run-123",
    checkpoint_storage=storage,
)

await workflow.run(
    message=10,
    checkpoint=checkpoint,
)

# Continue the same logical execution.
await workflow.run(
    responses={"approval-request": True},
    checkpoint=checkpoint,
)

latest_snapshot = checkpoint.head_snapshot

A one-shot invocation without a checkpoint remains stateless:

result = await workflow.run(message=10)

To resume after a process restart, load the checkpoint and pass it directly to the workflow:

storage = FileCheckpointStorage("./checkpoints")
checkpoint = await storage.load("factor-run-123")

workflow = build_workflow()
result = await workflow.run(checkpoint=checkpoint)

A checkpoint returned by a storage implementation should remain attached to that storage so subsequent snapshots continue to be persisted automatically.

The exact property names should be finalized in an ADR, but the intended public shape is:

class Checkpoint:
    checkpoint_id: str
    workflow_name: str
    graph_signature_hash: str
    snapshots: list[CheckpointSnapshot]
    head_snapshot_id: str | None
    checkpoint_storage: CheckpointStorage | None  # Runtime-only attachment

    @property
    def head_snapshot(self) -> CheckpointSnapshot | None: ...

    def get_snapshot(self, snapshot_id: str) -> CheckpointSnapshot: ...

    def restart_from(self, snapshot_id: str) -> None: ...

    async def commit_snapshot(
        self,
        *,
        state: dict[str, Any],
        metadata: dict[str, Any] | None = None,
    ) -> CheckpointSnapshot: ...


class CheckpointSnapshot:
    snapshot_id: str
    parent_snapshot_id: str | None
    timestamp: str
    state: dict[str, Any]
    metadata: dict[str, Any]


class Workflow:
    def create_checkpoint(
        self,
        *,
        checkpoint_id: str | None = None,
        checkpoint_storage: CheckpointStorage | None = None,
    ) -> Checkpoint: ...

    def run(
        self,
        message: Any | None = None,
        *,
        checkpoint: Checkpoint | None = None,
        responses: Mapping[str, Any] | None = None,
        stream: bool = False,
        ...,
    ) -> ...: ...

checkpoint_storage is attached to the checkpoint for automatic persistence but is not part of its serialized representation.

Checkpoint and snapshot model

A Checkpoint represents the developer-owned lifecycle and snapshot history of one logical workflow execution.

A CheckpointSnapshot represents the complete workflow state at one resumable boundary:

• initial input has been accepted;
• a superstep has completed;
• the workflow has paused; or
• the workflow has completed.

head_snapshot_id identifies the snapshot from which the next run will continue. For a newly created checkpoint it is None. After a snapshot is committed, the head advances to that snapshot.

All mutable execution state must be stored under four framework-owned keys in CheckpointSnapshot.state:

snapshot.state = {
    "workflow_state": {
        # Iteration count, run status, workflow-level shared state,
        # continuation cursor, and other workflow-wide state.
    },
    "executor_states": {
        # State keyed by executor ID.
        #
        # An AgentExecutor stores its AgentSession here rather than
        # retaining the session on the workflow-owned executor instance.
    },
    "edge_states": {
        # Delivery, fan-in, buffering, and other edge-runner state,
        # keyed by stable edge identity.
    },
    "messages": {
        "input": ...,
        "pending": ...,
        "output": ...,
    },
}

These are the only top-level state buckets:

• workflow_state contains workflow-wide state, including iteration count and run status.
• executor_states contains state owned by individual executors. For an AgentExecutor, this includes its AgentSession.
• edge_states contains state required to restore message delivery and edge behavior.
• messages.input contains the input that entered the workflow.
• messages.pending contains messages or requests that have not yet been fully processed, including pending external-input requests.
• messages.output contains outputs produced by the workflow up to that snapshot.

Framework additions should be made inside one of these buckets rather than adding more top-level snapshot-state fields.

Snapshot identity, lineage, timestamps, format version, and diagnostic metadata remain typed snapshot fields because they describe the snapshot rather than the workflow execution state.

Stateless workflow execution

Workflow should contain only the reusable workflow definition:

• graph topology and graph signature;
• executor definitions or factories;
• input/output metadata;
• execution configuration such as max_iterations.

Every call to run() should create an ephemeral runner hydrated from the checkpoint's head snapshot.

After the run returns, no mutable execution state may remain on the workflow. The supplied checkpoint is the only object through which execution state survives across calls.

A single workflow must support concurrent runs with different checkpoints. Concurrent use of the same checkpoint must be rejected to prevent lost snapshots and conflicting state updates.

Stateful executor instances are the main complication. Moving only Workflow._runner into a local variable is insufficient if executor objects continue to retain mutable fields. The implementation must either:

• create run-scoped executor instances; or
• move all executor state into CheckpointSnapshot.state["executor_states"].

AgentExecutor must similarly stop retaining its AgentSession on a workflow-owned executor instance. Nested agent sessions belong in the executor's snapshot state.

Restarting from an earlier snapshot

A checkpoint must make it easy to restart from any retained snapshot, not only the latest one:

checkpoint = await storage.load("factor-run-123")
checkpoint.restart_from(snapshot_id="snapshot-5")

await workflow.run(checkpoint=checkpoint)

restart_from() changes the checkpoint head but does not delete snapshots created after the selected snapshot.

The next committed snapshot uses the selected snapshot as its parent:

snapshot-4
    |
snapshot-5 -------- snapshot-6b
    |
snapshot-6 -------- snapshot-7

This preserves the original history while allowing a new execution branch to be created from snapshot-5.

Required behavior:

• restart_from() validates that the snapshot belongs to the checkpoint.
• Workflow graph compatibility is validated before execution resumes.
• The selected snapshot becomes head_snapshot.
• The next run restores exclusively from the selected snapshot's state.
• New snapshots set parent_snapshot_id to the current head.
• A successful snapshot commit advances head_snapshot_id.
• Existing descendant snapshots are retained until the developer explicitly deletes or prunes them.
• Storage persists both snapshot lineage and the checkpoint head.
• Loading a checkpoint restores its persisted head while still allowing another retained snapshot to be selected.

This makes the checkpoint a small snapshot graph rather than assuming that its history is always linear.

Committing and persisting snapshots

The workflow runner should only export its complete state. Snapshot creation, lineage, persistence, and head advancement belong to Checkpoint.commit_snapshot():

snapshot = await checkpoint.commit_snapshot(
    state=export_complete_run_state(),
    metadata={"boundary": "superstep_completed"},
)

Conceptually, commit_snapshot() performs:

class Checkpoint:
    async def commit_snapshot(
        self,
        *,
        state: dict[str, Any],
        metadata: dict[str, Any] | None = None,
    ) -> CheckpointSnapshot:
        previous_head = self.head_snapshot_id

        snapshot = CheckpointSnapshot(
            parent_snapshot_id=previous_head,
            state=state,
            metadata=metadata or {},
        )

        if self.checkpoint_storage is not None:
            await self.checkpoint_storage.commit_snapshot(
                self,
                snapshot,
                expected_head_snapshot_id=previous_head,
            )

        self.snapshots.append(snapshot)
        self.head_snapshot_id = snapshot.snapshot_id
        return snapshot

The actual implementation must synchronize commits so two callers cannot concurrently advance the same checkpoint.

commit_snapshot() is responsible for:

1. Capturing the current head.
2. Creating the snapshot and parent relationship.
3. Asking attached storage to persist the snapshot and new head atomically.
4. Appending the snapshot to in-memory history only after persistence succeeds.
5. Advancing the in-memory head.
6. Returning the committed snapshot.

The runner must not construct lineage, call storage directly, or update head_snapshot_id itself.

The lower-level storage contract can be:

class CheckpointStorage(Protocol):
    async def commit_snapshot(
        self,
        checkpoint: Checkpoint,
        snapshot: CheckpointSnapshot,
        *,
        expected_head_snapshot_id: str | None,
    ) -> None: ...

    async def load(self, checkpoint_id: str) -> Checkpoint: ...

    async def delete(self, checkpoint_id: str) -> None: ...

Persisting a snapshot must also persist the checkpoint's new head in the same logical operation. There must not be separate save_snapshot() and set_head() calls.

CheckpointStorage.commit_snapshot() must atomically:

1. Persist the new snapshot and its parent relationship.
2. Set the persisted checkpoint head to the new snapshot.
3. Validate that the persisted head still matches expected_head_snapshot_id.
4. Fail with a checkpoint-conflict error if another writer advanced the head.

For file storage, this can be implemented through atomic replacement of the checkpoint representation or an atomically replaced manifest containing the head. Database-backed implementations should update the snapshot and head in one transaction.

If persistence fails:

• the persisted head remains unchanged;
• the in-memory head remains unchanged;
• the candidate snapshot is not visible as a committed snapshot;
• the failure is surfaced to the caller.

A backend may temporarily leave an unreachable snapshot record after an interrupted write, but loading the checkpoint must only expose snapshots reachable through committed checkpoint metadata.

When no storage is attached, Checkpoint.commit_snapshot() appends the snapshot and advances the head in memory.

Checkpoint recovery remains at-least-once. The framework cannot transactionally roll back external side effects completed during a superstep, so this limitation must be documented.

Snapshot history and lifecycle ownership

The checkpoint has one stable checkpoint_id for its complete lifecycle. Individual supersteps create snapshot_id values rather than unrelated top-level checkpoint IDs.

This removes the need for:

await storage.get_latest(workflow_name=workflow.name)

The caller already owns the checkpoint and can inspect its head directly:

latest = checkpoint.head_snapshot

The developer owns:

• checkpoint identity;
• checkpoint retention;
• snapshot retention or pruning;
• head selection;
• deletion;
• persistence backend selection;
• restoration and branching.

Storage implementations may persist snapshots incrementally or load older snapshots lazily, but those implementation details must not change the public checkpoint model.

Adopt session-style msgspec serialization

Replace the current JSON plus pickle/base64 checkpoint encoding with the same approach used by FileSessionStore:

• A typed, versioned internal msgspec.Struct for Checkpoint.
• A typed, versioned internal msgspec.Struct for CheckpointSnapshot.
• msgspec JSON as the readable default.
• Optional msgspec MessagePack using the same model.
• Atomic file writes or transactional backend writes.
• Strict envelope and version validation.
• Explicit custom-state type registration.
• Clear errors for unsupported values and unknown versions.

The dynamic CheckpointSnapshot.state dictionary should use the same recursive codec and type registry as AgentSession.state. We should not maintain separate serialization systems for agent and workflow state.

Types registered through register_state_type should round-trip in both agent sessions and workflow snapshots. If that name is too agent-specific, it can be renamed to a neutral state-serialization registration API as part of the v2 break.

Remove the pickle-based checkpoint path, including:

• register_checkpoint_type;
• allowed_checkpoint_types;
• restricted-unpickler configuration;
• pickle/base64 marker handling.

New v2 writes must never silently fall back to pickle. If v1 checkpoints must survive the major-version upgrade, provide an explicit migration reader or offline migration path rather than enabling automatic legacy deserialization during normal v2 loads.

Other required updates

• Remove checkpoint storage from WorkflowBuilder.
• Replace checkpoint_id and checkpoint_storage arguments on Workflow.run() with checkpoint.
• Move the workflow-wide active-run guard to the checkpoint.
• Rework RunnerImpl, InProcRunnerContext, executor state, and edge-runner state so they are run-scoped.
• Move AgentExecutor's nested AgentSession into snapshot executor state.
• Update functional workflows and pipelines to use the same checkpoint model.
• Update nested workflows so child workflow state is represented in the parent's snapshot state.
• Update WorkflowAgent to store its workflow checkpoint through the surrounding AgentSession.state.
• Update hosting WorkflowState to store and retrieve checkpoints instead of maintaining a separate session_id -> checkpoint_id cursor.
• Update AG-UI, A2A, Foundry hosting, and other adapters that currently forward checkpoint_id or checkpoint_storage.
• Update in-memory, file, Cosmos, and other checkpoint storage implementations.
• Include checkpoint and snapshot identifiers in telemetry.
• Update checkpoint, human-in-the-loop, hosting, and orchestration samples.
• Add an ADR covering ownership, concurrency, storage ordering, snapshot retention, serialization, and v1 migration.

Acceptance criteria

• workflow.create_checkpoint() returns a caller-owned Checkpoint.
• workflow.run(..., checkpoint=checkpoint) restores from and updates that checkpoint.
• A checkpoint contains snapshots, and every snapshot contains the complete resumable state dictionary.
• Snapshot state has exactly four top-level buckets: workflow_state, executor_states, edge_states, and messages.
• Messages are divided into input, pending, and output.
• Workflow-wide values such as iteration count and run status are stored inside workflow_state.
• An AgentExecutor stores its AgentSession inside its entry in executor_states.
• A Workflow retains no mutable execution state between calls.
• Separate checkpoints can run concurrently against one workflow without state leakage.
• Concurrent use of the same checkpoint is rejected.
• Omitting a checkpoint produces an independent one-shot run.
• Reusing a checkpoint preserves all workflow, executor, edge, and message state.
• An entry snapshot and a snapshot after each completed superstep are created.
• Snapshot creation, parent assignment, persistence, and head advancement are encapsulated by Checkpoint.commit_snapshot().
• Workflow runners do not manipulate checkpoint lineage or storage directly.
• Attached storage persists every snapshot and its new checkpoint head as one atomic, conflict-checked operation.
• Storage failures are surfaced and leave the checkpoint at its last durable head.
• A stored checkpoint can resume against a rebuilt workflow in a new process.
• A caller can restart from any retained snapshot without deleting newer snapshots.
• Restarting from an earlier snapshot creates a new lineage branch.
• Each snapshot records its parent, and the checkpoint records its current head.
• Graph-signature compatibility is validated before restoration.
• JSON and MessagePack checkpoints round-trip registered framework and application state through msgspec.
• New checkpoint serialization contains no pickle payloads.
• Unsupported versions, incompatible graph signatures, corrupt data, and unregistered custom values fail clearly.
• Existing workflow integrations and samples use the new checkpoint-based API.
• The change is introduced only in v2; v1 does not gain a parallel lifecycle model.

Non-goals

• Introducing a separate WorkflowSession abstraction.
• Exactly-once execution of external side effects.
• Automatic snapshot retention or garbage collection.
• Treating checkpoint IDs as authorization boundaries.
• Silently loading legacy pickle-based checkpoints in the normal v2 code path.

Language/SDK

Python