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

[Oleg Solozobov (dev404ai)]

Motivation & Context

Closes #7831

FileCheckpointStorage.get_latest was built on list_checkpoints, which catches every exception per file, logs a warning and returns whatever it could read. A checkpoint that save accepted but load refuses therefore disappeared from the listing, and get_latest returned the newest of the survivors. A workflow resumed from earlier state after being told its newer state was saved, and the only trace was a logger.warning on a path nobody watches during recovery.

Two encoding rules make that reachable rather than theoretical. encode_checkpoint_value applies no allow list, so save accepts any state value. decode_checkpoint_value does apply one, so load can refuse the same checkpoint save accepted. Restricting deserialization is a deliberate security control and nothing here changes it: the problem is only that a refusal is currently converted into a plausible wrong answer.

Closed issue #3529 asked for the same thing from the other side. Its list_checkpoints row reads TODO: improve this API .get_latest(), error, human in the list. The issue closed once get_latest() existed; because get_latest() was built on list_checkpoints, the error half of that line stayed open. This closes it.

Description

get_latest now identifies the newest checkpoint from stored metadata and loads only that one.

workflow_name, timestamp and checkpoint_id are plain JSON in the stored file, so the newest checkpoint can be found without decoding any payload. Once found, it is loaded through load, which means an undecodable newest checkpoint raises WorkflowCheckpointException carrying the decoder's own actionable message naming allowed_checkpoint_types, instead of being skipped in favor of an older one.

Two alternatives were rejected on their merits.

Validating encodable types at save is the wrong layer. The allow list is a read-time policy configured per storage instance through allowed_checkpoint_types, so the writer cannot know the reader's policy, which may live in a different process. Enforcing it on write would break both the intent of the security model and existing workflows that widen the list on the reading side.

Making list_checkpoints raise is breaking, and it destroys the robustness its try/except exists to provide: one unreadable file would break every listing. list_checkpoints is therefore left exactly as it is, so nothing that lists checkpoints starts failing.

A side effect worth noting: get_latest no longer decodes every checkpoint in order to return one.

The same catch-and-warn shape exists in agent_framework_azure_cosmos._checkpoint_storage, whose get_latest also builds on its own listing. This pull request deliberately does not touch it, because that backend cannot be exercised here without an Azure account and shipping an unverified change to a backend I cannot run would be worse than scoping honestly.

Contribution Checklist

• The code builds clean without any errors or warnings
• I didn't break anyone
• I have added tests that prove my fix is effective or that my feature works

Four tests were added to python/packages/core/tests/workflow/test_checkpoint.py:

• the newest checkpoint is returned when everything decodes
• an undecodable newest checkpoint raises rather than falling back to an older one
• the empty case still returns None
• list_checkpoints still skips unreadable entries, so its tolerance is pinned rather than assumed

On main at d9d3fb62, with the tests applied and the fix reversed, the second of those fails with Failed: DID NOT RAISE WorkflowCheckpointException while the other three pass. With the fix applied, that file passes 48 of 48.

The full workflow suite reports 966 collected, 962 passed, 0 failed, 0 errors, 4 skipped in 42.30s:

uv run pytest packages/core/tests/workflow/

ruff check and ruff format --check are clean on both changed files. The change is 89 insertions and 5 deletions across two files.