Skip to content

Python: Add a typed schema for AgentExecutor checkpoint state #8201

Description

Summary

Add a public typed schema for the checkpoint state produced and restored by AgentExecutor.

Current behavior

The generic executor checkpoint hooks use dict[str, Any]:

async def on_checkpoint_save(self) -> dict[str, Any]: ...
async def on_checkpoint_restore(self, state: dict[str, Any]) -> None: ...

AgentExecutor.on_checkpoint_save() currently returns a mapping containing:

  • cache
  • full_conversation
  • agent_session
  • pending_agent_requests
  • pending_responses_to_agent

The payload is stored under WorkflowCheckpoint.state["_executor_state"][executor_id]. The shape is not currently represented by a public TypedDict or equivalent model.

Request

Please introduce and document a typed checkpoint-state contract for AgentExecutor, including:

  1. A public TypedDict or equivalent type for the serialized state.
  2. Typed handling for the serialized AgentSession payload.
  3. Clear compatibility/versioning guidance for adding or removing fields.
  4. Validation and useful errors for malformed checkpoint data during restore.
  5. Tests covering serialization, restoration, missing optional fields, and backward compatibility.
  6. Guidance for custom executors that implement their own checkpoint state.

Rationale

Applications that need to inspect, migrate, clean up, or associate executor checkpoint data with application sessions currently have to rely on an undocumented dict[str, Any] shape. A typed contract would improve static analysis and make checkpoint persistence safer to evolve without confusing internal state with a stable public API.

Activity

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Metadata

Metadata

Labels

agentsUsage: [Issues, PRs], Target: Single agentpythonUsage: [Issues, PRs], Target: Python

Projects

Milestone

No milestone

Relationships

None yet

Development

No branches or pull requests

Issue actions