Python: [Bug]: get_latest returns a stale checkpoint when the newest one cannot be decoded #7831

Description

A checkpoint that save accepts can be impossible to read back, and when that happens get_latest returns an older checkpoint instead of reporting the problem. A workflow then resumes from earlier state, having been told its newer state was saved.

Three behaviors combine to produce that, none of them wrong on its own. Encoding applies no allow list, so save accepts any state value, while decoding does apply one, so load can refuse the same checkpoint. Listing catches every exception per file, logs a warning and returns whatever it could read. And get_latest is built on list_checkpoints, so it inherits the partial view and picks the newest of the survivors.

The only signal is a logger.warning, on a code path nobody is watching during recovery.

Closed issue #3529 raised this API from the other side. Its list_checkpoints row reads TODO: improve this API .get_latest(), error, human in the list, and the issue closed once get_latest() existed. get_latest() was built on list_checkpoints, so the error half of that line is still open.

The recent work on the allow list is what makes the refusal path worth handling rather than hypothetical. #7791 registers built-in orchestration types and #7789 reports group chat envelope types being rejected on restore. Both widen what decode accepts, which is the right fix for those types. Neither changes what recovery does when decode refuses something anyway, and any application type outside the registered set still lands on that path.

The Azure Cosmos storage in agent_framework_azure_cosmos._checkpoint_storage uses the same catch-and-warn pattern, and its get_latest also builds on its own listing, so it appears to share the behavior. I have not verified that, because I cannot exercise it without an Azure account, and my pull request does not touch it.

I have a fix ready with tests and will open a pull request alongside this issue.

Code Sample

import asyncio
import tempfile
from dataclasses import dataclass

from agent_framework._workflows._checkpoint import FileCheckpointStorage, WorkflowCheckpoint


@dataclass
class AppState:
    stage: str


def checkpoint(state: dict, timestamp: str) -> WorkflowCheckpoint:
    return WorkflowCheckpoint(
        workflow_name="recovery-probe",
        graph_signature_hash="probe-signature",
        state=state,
        timestamp=timestamp,
    )


async def main() -> None:
    with tempfile.TemporaryDirectory() as tmp:
        storage = FileCheckpointStorage(storage_path=tmp)

        await storage.save(checkpoint({"stage": "step-1"}, "2026-08-16T10:00:00+00:00"))
        newer_id = await storage.save(checkpoint({"app": AppState(stage="step-2")}, "2026-08-16T11:00:00+00:00"))
        print(f"save of the newer checkpoint succeeded, id={newer_id}")

        latest = await storage.get_latest(workflow_name="recovery-probe")
        print(f"get_latest returns {latest.state if latest else None}")
        print(f"is it the newest save? {latest is not None and latest.checkpoint_id == newer_id}")


asyncio.run(main())

Output on main at d9d3fb62:

save of the newer checkpoint succeeded, id=48a6f608-b93e-4a71-acef-8692833a9e4f
get_latest returns {'stage': 'step-1'}
is it the newest save? False

Expected: get_latest either returns the newest checkpoint or raises. Returning an older one is the answer to a question nobody asked.

Error Messages / Stack Traces

The warning that precedes the wrong answer is the only trace the refusal leaves:

Failed to read checkpoint file /tmp/.../48a6f608-b93e-4a71-acef-8692833a9e4f.json:
Failed to decode pickled checkpoint data: Checkpoint deserialization blocked for
type '__main__:AppState'.

Package Versions

agent-framework-core: 1.13.0

Python Version

Python 3.13.7

Additional Context

Restricting deserialization by default is a deliberate security control, and per storage configuration through allowed_checkpoint_types is how an application widens it. Nothing about that needs to change. The problem is only that a refusal to decode is currently converted into a plausible wrong answer instead of an error.

workflow_name, timestamp and checkpoint_id are stored as plain JSON, so the newest checkpoint can be identified without decoding any payload. My pull request takes that route, which also means get_latest stops decoding every checkpoint in order to return one, and it leaves list_checkpoints tolerant exactly as it is today, so nothing that lists checkpoints starts failing.