Python: fix get_latest silently returning a stale checkpoint on decode failure

[pratik wayase (PratikWayase)]

Motivation & Context

FileCheckpointStorage.get_latest() was silently returning stale state. When the newest checkpoint on disk could not be decoded — for example, because its state payload referenced a type not in allowed_checkpoint_types — get_latest would fall back to the next-newest checkpoint it could read, and return that as if it were the latest. A workflow then resumed from earlier state, having been told its newer state was saved.

The root cause was that get_latest was built on 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 compose 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), read only the plain-JSON envelope — workflow_name, timestamp, checkpoint_id. No payload decoding happens in this phase. Files whose workflow_name doesn't match the requested workflow are skipped; any file that could belong to the requested workflow but is unreadable raises WorkflowCheckpointException.
  2. Post-scan validation. Re-stat every file 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 "changed during the scan; retry get_latest". The caller can retry.
  3. Decode only the winner. Decode the selected checkpoint from the dict already read in phase 1 — do not reopen the file via load(), which 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 should not break listing. The asymmetry is pinned by a new test (test_file_checkpoint_storage_list_checkpoints_still_skips_undecodable_entries). Issue Python: FileCheckpointStorage saves state it cannot restore: silent save/load asymmetry, get_latest invisibility, listing disagreement, and undocumented JSONDecodeError #8181 raises the broader consistency question of whether list_checkpoints should also surface decode errors; this PR does not take a position on that and leaves the door open.

  A small _decode_checkpoint() helper is extracted and 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 in order to return one — only the winner is decoded.
  • Concurrent writes during a get_latest call are detected and surfaced as a retryable error rather than silently returning a snapshot that may already be stale.
  • The selected checkpoint is decoded from the already-read dict, eliminating 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.

• What do you want reviewers to focus on?

  1. 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, the version tuple was reduced to (st_size, st_mtime_ns, st_ctime_ns) and the two intra-file os.fstat() comparisons (pre-read and post-read) were removed. 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 does not weaken that guard. Reviewers on Linux/macOS should sanity-check that the reduced tuple is still sufficient there.
  2. 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.
  3. 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: used to help understand the issue and the surrounding code.

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 fields in unrelated workflows can block recovery, and Windows mutation tests can pass without exercising scan validation.

2 open issues

• Validate duplicate keys only after identifying workflow ownership · New
• Close checkpoint read handle before delete or replace mutations · New

What changed in this PR

Fixes #7831 by making Python’s file checkpoint recovery raise when the newest checkpoint cannot be decoded, rather than silently returning older state.

Changes:

• Selects checkpoints from metadata, validates the scan, and decodes only the winner.
• Shares decoding logic with load() while preserving tolerant listing.
• Adds recovery, metadata, concurrency, and deserialization-policy tests.

File	Description
python/​packages/​core/​tests/​workflow/​test_checkpoint.py	Adds checkpoint recovery regression tests.
python/​packages/​core/​agent_framework/​_workflows/​_checkpoint.py	Implements metadata-first selection and snapshot validation.

_{Review effort: Balanced}