[pratik wayase (PratikWayase)]

Motivation & Context

This is a follow-up to #9485, which got closed under the stale-contribution rule while I was still working on Eduard van Valkenburg (@eavanvalkenburg)'s review feedback. Both of his comments are now addressed, with regression tests for each.

The bug itself: FileCheckpointStorage.get_latest() was silently returning stale state. When the newest checkpoint on disk couldn't be decoded (say, its state payload referenced a type not in allowed_checkpoint_types), get_latest would just fall back to the next-newest checkpoint it could read and return that as if it were the latest. A workflow would then resume from earlier state, believing its newer state was saved.

The root cause is that get_latest was built on top of list_checkpoints, which catches per-file decode failures, logs a warning, and returns whatever it could read. The only signal of the refusal was a logger.warning on a code path nobody watches during recovery. Issue #7831 describes the bug and the three behaviors that combine to produce it.

Description & Review Guide

What are the major changes?

get_latest no longer goes through list_checkpoints. It scans the storage directory itself in three phases:

1. Metadata-only scan. For each .json file under the storage root (recursively, with symlink/cycle protection), it reads only the plain-JSON envelope — workflow_name, timestamp, checkpoint_id. No payload decoding happens here. Files whose workflow_name doesn't match the requested workflow get skipped; any file that could belong to the requested workflow but is unreadable raises WorkflowCheckpointException.

2. Post-scan validation. Re-stat every file that was read (using Path.stat() consistently) and re-scan the directory tree. If anything changed during the scan — a file's mtime/size/ctime, the set of files, the set of directories, or the storage root itself — raise _CheckpointScanInvalidated. The outer retry loop catches it and retries up to 3 times with backoff.

3. Decode only the winner. Decode the selected checkpoint from the dict that was already read in phase 1 — do not reopen the file via load(), because that could decode a replacement rather than the validated snapshot. If decode fails, raise; do not fall back to an older checkpoint.

list_checkpoints() is intentionally left unchanged. It stays tolerant — one bad file shouldn't break listing. The asymmetry is pinned by a new test (test_file_checkpoint_storage_list_checkpoints_still_skips_undecodable_entries). Issue #8181 raises the broader question of whether list_checkpoints should also surface decode errors; I'm not taking a position on that here, just leaving the door open for a follow-up.

I also extracted a small _decode_checkpoint() helper shared between load() and get_latest() so both apply the same allowed_checkpoint_types policy. Behavior of load() is unchanged.

What is the impact of these changes?

• The exact bug from Python: [Bug]: get_latest returns a stale checkpoint when the newest one cannot be decoded #7831 is fixed: an undecodable newest checkpoint now raises instead of silently falling back.
• get_latest stops decoding every checkpoint just to return one — only the winner gets decoded.
• Concurrent writes/deletes during a get_latest call are detected and retried internally, instead of silently returning a snapshot that may already be stale, or failing recovery after a single scan.
• The selected checkpoint is decoded from the already-read dict, which closes a TOCTOU window where a concurrent writer could swap the file between selection and decode.
• list_checkpoints() behavior is unchanged — nothing that lists checkpoints starts failing.
• Files whose owning workflow can't be identified (unparseable JSON, missing/non-string workflow_name, or duplicate workflow_name at the root) are quarantined (skipped with a warning) so one malformed or legacy file can't permanently break recovery for every workflow sharing the storage root. Corruption in a file that is identified as belonging to the requested workflow still raises.

What do you want reviewers to focus on?

1. The two review comments from Eduard van Valkenburg (@eavanvalkenburg) on Python: fix get_latest silently returning a stale checkpoint on decode failure #9485 are addressed:

   • Deletion race (the blocker). FileNotFoundError during the initial read (between _checkpoint_paths() and stat()/open()) is now caught separately and re-raised as _CheckpointScanInvalidated, so the outer retry loop handles it. The post-loop re-stat catches FileNotFoundError the same way. Deleting even an unrelated workflow's checkpoint mid-scan no longer fails otherwise healthy recovery after one scan — the next attempt returns the surviving checkpoints. Pinned by test_file_checkpoint_storage_get_latest_retries_when_checkpoint_disappears_before_read.
   • Nested duplicate workflow_name. _make_duplicate_capturing_hook now returns separate root_duplicate_keys and all_duplicate_keys trackers. The ambiguous-ownership check uses only root_duplicate_keys, so a duplicated workflow_name inside metadata or state no longer quarantines a matching-workflow file as unowned. The nested duplicate is surfaced as a key-uniqueness violation once ownership is established. Pinned by test_file_checkpoint_storage_get_latest_surfaces_nested_duplicate_workflow_name. This closes the silent-stale-state recurrence that the prior review flagged.

2. The Windows-specific deviation from the original PR Python: raise instead of silently returning a stale checkpoint from get_latest #7832 design. On Windows, Path.stat() and os.fstat(fd) use different Win32 APIs (GetFileAttributesExW vs GetFileInformationByHandle) and can return different values for st_dev, st_ino, and even st_mtime_ns / st_ctime_ns on the same stable file. To make the scan work on Windows, I reduced the version tuple to (st_size, st_mtime_ns, st_ctime_ns) and removed the two intra-file os.fstat() comparisons (pre-read and post-read). The post-loop re-stat uses Path.stat() consistently and still detects concurrent writes. Symlink escapes are caught separately by _validate_file_path and the is_relative_to(root) check in _checkpoint_paths, so dropping st_dev doesn't weaken that guard. Reviewers on Linux/macOS should sanity-check that the reduced tuple is still sufficient there.

3. The decision to keep list_checkpoints tolerant. This is intentional and pinned by a test, but if maintainers want symmetry with get_latest (per Python: FileCheckpointStorage saves state it cannot restore: silent save/load asymmetry, get_latest invisibility, listing disagreement, and undocumented JSONDecodeError #8181), that's a follow-up PR — not this one.

4. The metadata-only scan and checkpoint_id validation. checkpoint_id must be a string and must resolve to the exact file being read (_validate_file_path(checkpoint_id) != file_path.resolve() → raise). This catches redirected-ID attacks where a file's metadata claims to be a different checkpoint. Top-level __pickled__ / __type__ markers are rejected because they could hide different metadata.

Related Issue

Fixes #7831

AI Assistance

• This is an AI-assisted contribution. I reviewed, understood, and verified all submitted content and accept responsibility for it.

AI assistance details: AI assistance was used only to help understand the issue and the surrounding code. All implementation, tests, and this PR description were written by me.

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 links to an agreed issue with no competing open PR, or the Related Issue section documents a trivial-change or repository-automation exception.
• This is not a breaking change. get_latest previously returned a stale checkpoint on decode failure of the newest; it now raises. Callers that depended on the silent-fallback behavior will see a new exception. No list_checkpoints / load / save / delete behavior changes.

[Copilot]

🟡 Changes recommended

Duplicate-key misclassification and skipped validation of malformed files can silently return stale checkpoints, while a regression test can fail with RecursionError on supported Python versions.

5 open issues

• Identify root duplicates from the final object_pairs_hook callback · New
• Track file versions before parsing checkpoint contents · New
• Avoid recursive Path.resolve calls in the stat patch · New
• Retry scans when queued directories disappear · New
• Filter foreign workflows before validating envelope markers · New

What changed in this PR

This PR addresses #7831 by changing Python’s FileCheckpointStorage.get_latest() to select checkpoints from stored metadata and surface decode failures rather than silently return older state.

Changes:

• Selects and decodes only the newest matching checkpoint, sharing decoding logic with load().
• Adds snapshot validation and bounded retries for concurrent filesystem changes.
• Adds regression coverage for decode failures, ownership validation, concurrent changes, and unchanged tolerant listing behavior.

File	Description
python/​packages/​core/​tests/​workflow/​test_checkpoint.py	Adds checkpoint-selection, corruption, and concurrency regression tests.
python/​packages/​core/​agent_framework/​_workflows/​_checkpoint.py	Implements metadata-based selection, snapshot validation, retries, and shared decoding.

_{Review effort: Balanced}

[Eduard van Valkenburg (eavanvalkenburg)]

pratik wayase (@PratikWayase) Checkpoint discovery checks only the entry name before treating it as a checkpoint candidate. A valid child store located at a directory such as root/archive.json/ is therefore treated as a checkpoint file; the later regular-file guard raises WorkflowCheckpointException and blocks recovery of healthy checkpoints in the parent store. Please skip directory entries without recursing into them, and extend the child-storage regression coverage with a directory whose name ends in .json.