[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}

[Jose Alvarez (jpalvarezl)]

The unconditional failure here means one malformed or legacy JSON file whose workflow owner cannot be identified makes get_latest() fail indefinitely for every workflow sharing this storage root, even when that workflow's checkpoints are healthy. Please isolate or quarantine unowned files so they cannot permanently poison unrelated workflows, while preserving an explicit failure for corrupt checkpoints known to belong to the requested workflow. Please also add shared-root regression coverage.

[Jose Alvarez (jpalvarezl)]

This invalidates the scan for any membership change under the shared root, including an atomic save for a different workflow. Because get_latest() immediately propagates the exception, ordinary concurrent checkpointing can fail unrelated recovery instead of retrying the optimistic read. Please add bounded internal retries for transient scan changes, or otherwise narrow invalidation without missing a newly written matching checkpoint, and cover a concurrent foreign-workflow save.