Python: raise instead of silently returning a stale checkpoint from get_latest

[Oleg Solozobov (dev404ai)]

Motivation & Context

Before this change, FileCheckpointStorage.get_latest() selected from the results of list_checkpoints(), which logs read errors and skips the affected files. If the newest saved checkpoint could not be decoded, recovery silently returned an older checkpoint or None.

This is reachable with the existing reader security policy: save() can encode a value that a later reader refuses to deserialize under allowed_checkpoint_types. The review also identified a metadata variant: an invalid timestamp can be saved successfully, then skipped during selection. Both cases must surface as recovery errors rather than silently discard newer state.

Description & Review Guide

• What are the major changes? Select the latest checkpoint from its stored workflow_name, timestamp and checkpoint_id fields, validate the selection, and decode only that checkpoint. Read and decode in one worker thread, using the same JSON captured during selection and a shared private helper that also serves load(). This preserves the reader's security policy without reopening a path that could now contain different data.
• What is the impact of these changes? Invalid matching metadata and an undecodable latest payload raise WorkflowCheckpointException, preserving the file context or the decoder's actionable error. ID redirection, duplicate JSON keys, root pickle envelopes, unreadable files, non-regular checkpoint paths, and observed changes during a scan also raise instead of permitting silent fallback.
• What do you want reviewers to focus on? The recovery error policy, compatibility for directories shared by several workflows, nested checkpoint paths and event-loop responsiveness, the limits of concurrent-change detection, and preservation of reader-side deserialization controls.

Unreadable JSON or a file without an identifiable workflow blocks selection for every workflow sharing the directory, because it might hide the requested workflow's newest checkpoint. Invalid selection fields for a clearly different workflow are ignored. Older payloads are not decoded merely to find the newest checkpoint.

The scan includes nested directories, so checkpoint IDs accepted by save() and load() remain discoverable. Readable directories without checkpoint files do not cause an error. Directory aliases within the storage root are visited once, including links to ancestors; paths resolving outside the root are rejected. Temporary .json.tmp files remain ignored, and existing empty and relative checkpoint ID aliases remain supported.

File versions are checked before and after reading, then checked again with file membership and the versions of every visited directory before the scan is accepted. These checks rely on filesystem change reporting and atomic writers such as the existing save(). They detect observed concurrent changes and allow the caller to retry; they do not provide transactional locking. A write after validation belongs to a later snapshot. Equal timestamps retain directory iteration order.

Two alternatives would change the wrong contract. Enforcing the reader's allowed types during save() would prevent a writer and a later reader from using different policies. Making list_checkpoints() raise would remove its existing tolerance. This change preserves both behaviors while making recovery errors explicit. Public signatures, the stored format, write behavior and other storage backends are unchanged. The Azure Cosmos backend remains outside this PR and was not tested locally.

Regression tests cover the original undecodable-payload failure, malformed timestamps with and without an older checkpoint, missing or invalid selection fields, ID redirection, mixed naive/aware dates, JSON ambiguity, file errors, concurrent writers, and decoding the selected snapshot once. Compatibility tests cover tolerant listing, older undecodable payloads, allowed custom types, timestamp offsets, ID aliases, temporary files, nested paths, shared file names across directories, directory links and relative storage roots. A synchronized regression requires another async task to run while decoding is paused; it fails if decoding returns to the event-loop thread. Additional tests cover nested corruption, access failures and changes during both scans.

Local verification on macOS with Python 3.13.7: 214 checkpoint tests passed. The complete workflow test directory produced 1046 passed, 2 skipped and 2 xfailed, with two warnings from unchanged tests. Ruff lint and formatting passed; scoped Pyright, mypy, Pyrefly, ty and Zuban checks passed. Linux, Windows and Azure integrations were not run locally. These results are from the local follow-up on b5d9ec7; they do not claim an upstream CI pass or verification against current main.

Related Issue

Closes #7831

Contribution Checklist

• The code builds clean without any errors or warnings
• All unit tests pass, and I have added new tests where possible
• The PR follows the Contribution Guidelines
• This PR is linked to an issue and there is no other open PR for this issue (see Related Issue above).
• This is not a breaking change. If it is a breaking change, add the breaking change label (or add "[BREAKING]" to the title prefix, before or after any language prefix) — a workflow keeps the label and title prefix in sync automatically.

The complete build and unit-test checklist items remain unchecked pending upstream CI. The non-breaking checkbox remains unchecked for maintainer review of the stricter recovery error policy: unreadable or unidentified JSON now stops recovery instead of being silently skipped. The flat-directory restriction has been removed.

[Copilot]

🟡 Changes recommended

The implementation introduces an inconsistent flat-directory contract and performs synchronous decoding on the event loop.

Once you've addressed the issues Copilot identified, you can request another Copilot review.

Pull request overview

Updates file-backed checkpoint recovery to raise when the newest checkpoint cannot be decoded instead of returning stale state.

Changes:

• Selects the latest checkpoint from JSON metadata before decoding.
• Adds filesystem consistency and corruption checks.
• Expands tests for decoding, metadata, concurrency, and filesystem edge cases.

File summaries

File	Description
python/packages/core/agent_framework/_workflows/_checkpoint.py	Implements metadata-based latest-checkpoint selection and validation.
python/packages/core/tests/workflow/test_checkpoint.py	Adds comprehensive get_latest behavior tests.

Review details

• Files reviewed: 2/2 changed files
• Comments generated: 3
• Review effort level: Balanced

---

💡 Add a code-review agent skill for context-aware, tailored reviews. Learn more in the docs.